packages feed

pgmq-config 0.4.0.1 → 0.5.0.0

raw patch · 10 files changed

+1257/−241 lines, 10 filesdep +aesondep +bytestringdep +postgresql-libpqdep ~pgmq-coredep ~pgmq-effectfuldep ~pgmq-hasql

Dependencies added: aeson, bytestring, postgresql-libpq

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

Files

CHANGELOG.md view
@@ -1,5 +1,80 @@ # Revision history for pgmq-config +## 0.5.0.0 -- 2026-08-06++The reconciler told three lies about what it had done, and could be taken down at startup+by a queue it does not manage. This release fixes all four and states the real contract.++### Breaking Changes++* `ReconcileAction` gains two constructors, and one existing constructor becomes+  reachable. Exhaustive matchers and consumers that count skips need updating.+  * `UpdatedNotifyThrottle !QueueName !Int32 !Int32` — the declared throttle interval+    differed from the stored row and was applied. Fields: queue, observed, declared.+  * `DetectedQueueTypeDrift !QueueName !QueueType !ObservedQueueType` — the queue's+    observed shape contradicts the declared one. This replaces `SkippedQueue` for the+    queue it concerns, so the report still carries exactly one queue-existence action per+    declared config.+  * `SkippedFifoIndex` was unreachable dead code and is now actually emitted.++### New Features++* `ObservedQueueType` (`ObservedStandard`, `ObservedUnlogged`, `ObservedPartitioned`) —+  the three-way queue shape `pgmq.list_queues()` actually reports.+* `defaultThrottleMs :: Int32` — the 250 ms interval pgmq applies when none is given. A+  `NotifyConfig` whose `throttleMs` is `Nothing` means "use this value".++### Bug Fixes++* A queue created by another client under a name `parseQueueName` rejects no longer fails+  your application's startup. `ensureQueues` snapshotted existing queues through the typed+  `listQueues`, whose decoder re-validates every name read back, so one foreign row — say+  `billing-events` — made reconciliation throw at boot. Someone else's queue became your+  boot failure. It now snapshots through `listQueuesUnvalidated` and matches declared+  names textually; foreign rows are simply queues it does not manage, consistent with its+  additive contract.+* `CreatedFifoIndex` is no longer reported when no index was created. The reconciler+  called `create_fifo_index` unconditionally and always reported creation. It now consults+  a `pg_indexes` snapshot, skips the call when the index is present, and reports which of+  the two happened.+* A declared notify throttle that differs from the stored row is now applied instead of+  silently ignored. The snapshot kept only queue names, so drift was invisible. It now+  carries intervals, and a difference is written via `pgmq.update_notify_insert` and+  reported as `UpdatedNotifyThrottle`. `Nothing` and a stored 250 compare equal, so a+  defaulted config does not flap, and an unchanged interval is still left strictly alone+  rather than re-enabled — re-enabling resets `last_notified_at`.+* A declared queue type contradicting the live queue is now reported as+  `DetectedQueueTypeDrift` rather than the misleading `SkippedQueue`. Nothing is mutated:+  converting a queue's type means dropping and recreating it, destroying its messages,+  which a startup reconciler must never do. Partition interval and retention are not+  drift-checked, because `pgmq.list_queues` does not report them.+* `withNotifyInsert Nothing` no longer fails reconciliation on every startup. It reached+  `pgmq.enable_notify_insert` as a SQL NULL throttle, raising SQLSTATE 23502; since+  reconciliation is not one transaction, this failed application startup repeatedly and+  permanently for any config using the default throttle. Fixed in the pgmq-hasql+  statement, which now coalesces to 250 ms.++### Documentation++* The package promised to "ensure all queues exist with the desired settings" and that+  "all operations are idempotent", neither of which described what the reconciler does.+  `ensureQueues`' Haddock is now the canonical statement of the contract: additive+  reconciliation, one documented mutation of existing state (throttle drift, with its+  `last_notified_at` side effect named), queue-type drift reported rather than repaired,+  partition settings deliberately unchecked, and the concurrent multi-replica caveat —+  SQLSTATE 42710 on stock upstream 1.11.0 extension installs, race-free on pgmq-migration+  installs via migration `0003`. `ensureQueuesReport` and both `Eff` twins point at that+  one description so the four copies cannot drift. Haddock coverage on the public modules+  is now 100%. The rationale is in `docs/design/018-reconciliation-contract.md`.++### Other Changes++* The duplicated `Session` and `Effectful` reconciliation logic is now a single+  backend-agnostic core in the internal `Pgmq.Config.Reconcile` module, parameterized over+  a `ReconcileOps` record; the public modules are thin adapters over it.+* Bumped `pgmq-core`, `pgmq-hasql`, and `pgmq-effectful` dependency bounds to+  `>=0.5 && <0.6`.+ ## 0.4.0.1 -- 2026-07-14  * Version bump only — coordinated release with pgmq-migration 0.4.0.1.
pgmq-config.cabal view
@@ -1,15 +1,23 @@ cabal-version:   3.4 name:            pgmq-config-version:         0.4.0.1+version:         0.5.0.0 synopsis:   Declarative queue configuration for PGMQ (PostgreSQL Message Queue)  description:   A declarative DSL for configuring pgmq queues. Define your queue   topology as Haskell values and call a single function at startup-  to ensure all queues exist with the desired settings. Supports-  standard, unlogged, and partitioned queues, notification setup,-  FIFO indexes, and topic bindings. All operations are idempotent.+  to create whatever is missing: standard, unlogged, and partitioned+  queues, insert notifications, FIFO indexes, and topic bindings.+  .+  Reconciliation is additive and convergent. It never drops, converts,+  or disables anything, and leaves queues that are absent from the+  config untouched; a second run against an unchanged config issues no+  mutations at all. The single exception is a declared notification+  throttle interval that differs from the stored one, which is updated+  in place. A queue whose observed type contradicts the declared one is+  reported as drift rather than repaired, since converting a queue's+  type would mean destroying its messages.  homepage:        https://github.com/shinzui/pgmq-hs license:         MIT@@ -38,6 +46,7 @@     Pgmq.Config     Pgmq.Config.Types +  other-modules:      Pgmq.Config.Reconcile   default-extensions:     DeriveGeneric     DuplicateRecordFields@@ -54,15 +63,15 @@     , hasql         ^>=1.10     , hasql-pool    ^>=1.4     , lens          ^>=5.3-    , pgmq-core     >=0.4   && <0.5-    , pgmq-hasql    >=0.4   && <0.5+    , pgmq-core     >=0.5   && <0.6+    , pgmq-hasql    >=0.5   && <0.6     , text          ^>=2.1    if flag(effectful)     exposed-modules: Pgmq.Config.Effectful     build-depends:       , effectful-core  ^>=2.5 || ^>=2.6-      , pgmq-effectful  >=0.4  && <0.5+      , pgmq-effectful  >=0.5  && <0.6    hs-source-dirs:     src   default-language:   GHC2024@@ -77,6 +86,8 @@   other-modules:     ConfigSpec     EphemeralDb+    ForeignQueueSpec+    NotifyCrashSpec    default-extensions:     DeriveGeneric@@ -88,19 +99,31 @@     OverloadedStrings    build-depends:-    , base            >=4.18  && <5-    , ephemeral-pg    >=0.2.1-    , generic-lens    ^>=2.2  || ^>=2.3+    , aeson             ^>=2.2+    , base              >=4.18   && <5+    , bytestring        >=0.11   && <0.13+    , ephemeral-pg      >=0.2.1+    , generic-lens      ^>=2.2   || ^>=2.3     , hasql-    , hasql-pool      ^>=1.4-    , lens            ^>=5.3+    , hasql-pool        ^>=1.4+    , lens              ^>=5.3     , pg-migrate     , pgmq-config     , pgmq-core     , pgmq-hasql     , pgmq-migration-    , random          ^>=1.2-    , tasty           ^>=1.5-    , tasty-hunit     ^>=0.10+    , postgresql-libpq  >=0.10.1 && <0.12+    , random            ^>=1.2+    , tasty             ^>=1.5+    , tasty-hunit       ^>=0.10     , text-    , time            ^>=1.14+    , time              ^>=1.14++  -- ForeignQueueSpec asserts that both reconciler backends use the lenient+  -- queue listing. The effect-backed half only exists when the library's+  -- effectful flag is on, so it is compiled in behind the same flag.+  if flag(effectful)+    cpp-options:   -DPGMQ_EFFECTFUL+    build-depends:+      , effectful-core  ^>=2.5 || ^>=2.6+      , pgmq-effectful  >=0.5  && <0.6
src/Pgmq/Config.hs view
@@ -1,3 +1,16 @@+-- | Declare a pgmq queue topology as Haskell values and create whatever is+-- missing, in one call at application startup.+--+-- Build configs with 'standardQueue', 'unloggedQueue', or 'partitionedQueue'+-- and refine them with 'withNotifyInsert', 'withFifoIndex', and+-- 'withTopicBinding'; then hand the list to 'ensureQueues' (or+-- 'ensureQueuesWithPool'), or to 'ensureQueuesReport' when you want to see what+-- was done. 'ensureQueues' documents the reconciliation contract in full —+-- what is created, the one case in which existing state is mutated, what is+-- reported as drift instead of repaired, and the concurrent-startup caveat.+--+-- The same reconciler is available over the @Pgmq@ effect in+-- "Pgmq.Config.Effectful". module Pgmq.Config   ( -- * Queue Configuration Types     QueueConfig (..),@@ -21,34 +34,88 @@      -- * Reconciliation with Report     ReconcileAction (..),+    ObservedQueueType (..),     ensureQueuesReport,++    -- * Defaults+    defaultThrottleMs,   ) where -import Control.Lens ((^.))-import Data.Generics.Labels ()-import Data.Set qualified as Set-import Data.Text qualified as T import Hasql.Pool qualified as Pool import Hasql.Session (Session)+import Pgmq.Config.Reconcile (ReconcileOps (..), ensureQueuesReportWith) import Pgmq.Config.Types import Pgmq.Hasql.Sessions qualified as Sessions-import Pgmq.Hasql.Statements.Types qualified as StmtTypes-import Pgmq.Types-  ( QueueName,-    TopicPattern,-    queueNameToText,-    topicPatternToText,-  ) --- | Ensure all declared queues exist with the desired settings.+-- | The 'Session'-backed wiring of the reconciliation operations.+sessionOps :: ReconcileOps Session+sessionOps =+  ReconcileOps+    { listQueuesUnvalidated = Sessions.listQueuesUnvalidated,+      listTopicBindings = Sessions.listTopicBindings,+      listNotifyInsertThrottles = Sessions.listNotifyInsertThrottles,+      createQueue = Sessions.createQueue,+      createUnloggedQueue = Sessions.createUnloggedQueue,+      createPartitionedQueue = Sessions.createPartitionedQueue,+      enableNotifyInsert = Sessions.enableNotifyInsert,+      createFifoIndex = Sessions.createFifoIndex,+      bindTopic = Sessions.bindTopic,+      listFifoIndexQueueNames = Sessions.listFifoIndexQueueNames,+      updateNotifyInsert = Sessions.updateNotifyInsert+    }++-- | Create whatever the declared configs call for that does not exist yet. ----- 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.+-- Reconciliation is /additive/. It snapshots existing queues, topic bindings,+-- notification throttles, and FIFO indexes with four read-only queries, then+-- issues mutating calls only for the pieces that are missing: it creates+-- queues, enables insert notifications, creates FIFO indexes, and binds topic+-- patterns. It never drops, converts, or disables anything, and a queue that+-- exists in the database but not in the config is left completely alone. ----- Operations are additive only: queues not in the config are left untouched.+-- There is exactly one exception, and it is deliberate: if a config declares a+-- notification throttle interval that differs from the one stored in the+-- database, the stored value is updated in place via+-- @pgmq.update_notify_insert@ and reported as+-- 'Pgmq.Config.Types.UpdatedNotifyThrottle'. That update also resets the+-- throttle's @last_notified_at@ to the epoch, so the next insert on that queue+-- raises a notification immediately — once, after a genuine configuration+-- change. A throttle whose interval already matches is not touched at all; in+-- particular it is not re-enabled, because re-enabling would reset the same+-- timestamp on every startup. @throttleMs = Nothing@ compares equal to+-- 'defaultThrottleMs', so a defaulted config does not flap.+--+-- What is /not/ reconciled:+--+-- * A queue whose observed shape (standard, unlogged, or partitioned)+--   contradicts the declared one is reported as+--   'Pgmq.Config.Types.DetectedQueueTypeDrift' and nothing is mutated.+--   Changing a queue's type means dropping and recreating it, destroying every+--   message it holds; that is an operator's decision, not a startup task.+--+-- * A partitioned queue's partition interval and retention interval are never+--   compared, because @pgmq.list_queues()@ does not report them. Only the+--   three-way shape is checked.+--+-- Calling this on every application startup is the intended usage, and a second+-- run against an unchanged config issues no mutations at all. One caveat about+-- /concurrent/ startups: queue creation and topic binding are serialized+-- server-side and converge, but enabling insert notifications on a+-- brand-new queue from two replicas at once can fail one of them with SQLSTATE+-- 42710 (@duplicate_object@) when pgmq was installed as the stock upstream+-- 1.11.0 extension. Databases installed through this repository's+-- @pgmq-migration@ package are free of that race — its migration+-- @0003-notify-crash-safety-and-locking.sql@ takes a per-queue advisory lock+-- inside @enable_notify_insert@. On an extension install, either retry the+-- reconcile (every operation is convergent, so a retry succeeds) or serialize+-- startup reconciliation across replicas. FIFO index creation has the same+-- narrow shape of race and is likewise harmless on retry, since the underlying+-- statement is @CREATE INDEX IF NOT EXISTS@.+--+-- Reconciliation is not wrapped in a transaction: each call autocommits, so a+-- failure part-way leaves the work done so far in place and the next run+-- continues from there. ensureQueues :: [QueueConfig] -> Session () ensureQueues configs = () <$ ensureQueuesReport configs @@ -57,96 +124,15 @@ ensureQueuesWithPool pool configs =   Pool.use pool (ensureQueues configs) --- | Like 'ensureQueues', but returns a report of actions taken.--- Queries existing state first and skips operations that are already satisfied.+-- | Like 'ensureQueues', but returns a report of what was done.+--+-- The contract is identical — see 'ensureQueues' for the full description of+-- what is and is not reconciled, and for the concurrent-startup caveat. The+-- report contains one action per decision the reconciler made, in declaration+-- order: exactly one queue-existence action per config+-- ('Pgmq.Config.Types.CreatedQueue', 'Pgmq.Config.Types.SkippedQueue', or+-- 'Pgmq.Config.Types.DetectedQueueTypeDrift'), then the notify, FIFO, and+-- topic-binding actions for that config. Every @Skipped@ action means no+-- statement was issued. ensureQueuesReport :: [QueueConfig] -> Session [ReconcileAction]-ensureQueuesReport configs = do-  existingQueues <- Sessions.listQueues-  existingBindings <- Sessions.listTopicBindings-  existingThrottles <- Sessions.listNotifyInsertThrottles--  let existingQueueNames = Set.fromList (map (\q -> q ^. #name) existingQueues)-      existingBindingSet =-        Set.fromList-          [ (b ^. #bindingQueueName, topicPatternToText (b ^. #bindingPattern))-          | b <- existingBindings-          ]-      existingNotifySet = Set.fromList (map (\t -> t ^. #throttleQueueName) existingThrottles)--  concat <$> traverse (reconcileQueue existingQueueNames existingBindingSet existingNotifySet) configs---- | Reconcile a single queue config against existing state, returning actions taken.-reconcileQueue ::-  Set.Set QueueName ->-  Set.Set (T.Text, T.Text) ->-  Set.Set T.Text ->-  QueueConfig ->-  Session [ReconcileAction]-reconcileQueue existingQueues existingBindings existingNotify cfg = do-  let qn = cfg ^. #queueName-      qnText = queueNameToText qn--  -- Queue creation-  queueAction <--    if Set.member qn existingQueues-      then pure [SkippedQueue qn]-      else do-        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-                }-        pure [CreatedQueue qn (cfg ^. #queueType)]--  -- Notification-  notifyAction <- case cfg ^. #notifyInsert of-    Nothing -> pure []-    Just nc ->-      if Set.member qnText existingNotify-        then pure [SkippedNotify qn]-        else do-          Sessions.enableNotifyInsert-            StmtTypes.EnableNotifyInsert-              { queueName = qn,-                throttleIntervalMs = nc ^. #throttleMs-              }-          pure [EnabledNotify qn (nc ^. #throttleMs)]--  -- FIFO index — no way to query if index exists, so always apply (idempotent)-  fifoAction <--    if cfg ^. #fifoIndex-      then do-        Sessions.createFifoIndex qn-        pure [CreatedFifoIndex qn]-      else pure []--  -- Topic bindings-  bindingActions <- concat <$> traverse (reconcileBinding qn qnText existingBindings) (cfg ^. #topicBindings)--  pure (queueAction ++ notifyAction ++ fifoAction ++ bindingActions)---- | Reconcile a single topic binding.-reconcileBinding ::-  QueueName ->-  T.Text ->-  Set.Set (T.Text, T.Text) ->-  TopicPattern ->-  Session [ReconcileAction]-reconcileBinding qn qnText existingBindings pat =-  let patText = topicPatternToText pat-   in if Set.member (qnText, patText) existingBindings-        then pure [SkippedTopicBinding qn pat]-        else do-          Sessions.bindTopic-            StmtTypes.BindTopic-              { topicPattern = pat,-                queueName = qn-              }-          pure [BoundTopic qn pat]+ensureQueuesReport = ensureQueuesReportWith sessionOps
src/Pgmq/Config/Effectful.hs view
@@ -1,3 +1,10 @@+-- | The reconciler of "Pgmq.Config", run over the @Pgmq@ effect from+-- pgmq-effectful instead of a @Hasql.Session.Session@.+--+-- Declaration types and the report type live in "Pgmq.Config.Types"; the+-- reconciliation contract is documented on 'Pgmq.Config.ensureQueues'. This+-- module is only built when the package's @effectful@ flag is on (it is by+-- default). module Pgmq.Config.Effectful   ( -- * Reconciliation     ensureQueuesEff,@@ -7,119 +14,50 @@   ) where -import Control.Lens ((^.))-import Data.Generics.Labels ()-import Data.Set qualified as Set-import Data.Text qualified as T import Effectful (Eff, (:>))+import Pgmq.Config.Reconcile (ReconcileOps (..), ensureQueuesReportWith) import Pgmq.Config.Types import Pgmq.Effectful.Effect qualified as Eff-import Pgmq.Hasql.Statements.Types qualified as StmtTypes-import Pgmq.Types-  ( QueueName,-    TopicPattern,-    queueNameToText,-    topicPatternToText,-  ) --- | Ensure all declared queues exist using the Pgmq effect.+-- | The @Pgmq@-effect-backed wiring of the reconciliation operations.+effectfulOps :: (Eff.Pgmq :> es) => ReconcileOps (Eff es)+effectfulOps =+  ReconcileOps+    { listQueuesUnvalidated = Eff.listQueuesUnvalidated,+      listTopicBindings = Eff.listTopicBindings,+      listNotifyInsertThrottles = Eff.listNotifyInsertThrottles,+      createQueue = Eff.createQueue,+      createUnloggedQueue = Eff.createUnloggedQueue,+      createPartitionedQueue = Eff.createPartitionedQueue,+      enableNotifyInsert = Eff.enableNotifyInsert,+      createFifoIndex = Eff.createFifoIndex,+      bindTopic = Eff.bindTopic,+      listFifoIndexQueueNames = Eff.listFifoIndexQueueNames,+      updateNotifyInsert = Eff.updateNotifyInsert+    }++-- | Create whatever the declared configs call for that does not exist yet,+-- through 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.+-- This runs the very same reconciler as 'Pgmq.Config.ensureQueues', over the+-- effect instead of a @Session@, so the contract is identical and is documented+-- once, there. In short: reconciliation is additive — it creates missing+-- queues, notification settings, FIFO indexes, and topic bindings, and never+-- drops or converts anything — with one deliberate exception, a declared+-- notification throttle interval that differs from the stored one, which is+-- updated in place. Queue-type drift is reported, not repaired. ----- Operations are additive only: queues not in the config are left untouched.+-- Read 'Pgmq.Config.ensureQueues' before relying on this at startup: it covers+-- the throttle update's @last_notified_at@ side effect, what is deliberately+-- left unchecked, and the concurrent multi-replica caveat (SQLSTATE 42710 on+-- stock upstream-1.11.0 extension installs). ensureQueuesEff :: (Eff.Pgmq :> es) => [QueueConfig] -> Eff es () ensureQueuesEff configs = () <$ ensureQueuesReportEff configs --- | Like 'ensureQueuesEff', but returns a report of actions taken.+-- | Like 'ensureQueuesEff', but returns a report of what was done.+--+-- Same report shape as 'Pgmq.Config.ensureQueuesReport': one action per+-- decision, one queue-existence action per config, and every @Skipped@ action+-- meaning no statement was issued. ensureQueuesReportEff :: (Eff.Pgmq :> es) => [QueueConfig] -> Eff es [ReconcileAction]-ensureQueuesReportEff configs = do-  existingQueues <- Eff.listQueues-  existingBindings <- Eff.listTopicBindings-  existingThrottles <- Eff.listNotifyInsertThrottles--  let existingQueueNames = Set.fromList (map (\q -> q ^. #name) existingQueues)-      existingBindingSet =-        Set.fromList-          [ (b ^. #bindingQueueName, topicPatternToText (b ^. #bindingPattern))-          | b <- existingBindings-          ]-      existingNotifySet = Set.fromList (map (\t -> t ^. #throttleQueueName) existingThrottles)--  concat <$> traverse (reconcileQueueEff existingQueueNames existingBindingSet existingNotifySet) configs---- | Reconcile a single queue config against existing state.-reconcileQueueEff ::-  (Eff.Pgmq :> es) =>-  Set.Set QueueName ->-  Set.Set (T.Text, T.Text) ->-  Set.Set T.Text ->-  QueueConfig ->-  Eff es [ReconcileAction]-reconcileQueueEff existingQueues existingBindings existingNotify cfg = do-  let qn = cfg ^. #queueName-      qnText = queueNameToText qn--  queueAction <--    if Set.member qn existingQueues-      then pure [SkippedQueue qn]-      else do-        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-                }-        pure [CreatedQueue qn (cfg ^. #queueType)]--  notifyAction <- case cfg ^. #notifyInsert of-    Nothing -> pure []-    Just nc ->-      if Set.member qnText existingNotify-        then pure [SkippedNotify qn]-        else do-          Eff.enableNotifyInsert-            StmtTypes.EnableNotifyInsert-              { queueName = qn,-                throttleIntervalMs = nc ^. #throttleMs-              }-          pure [EnabledNotify qn (nc ^. #throttleMs)]--  fifoAction <--    if cfg ^. #fifoIndex-      then do-        Eff.createFifoIndex qn-        pure [CreatedFifoIndex qn]-      else pure []--  bindingActions <- concat <$> traverse (reconcileBindingEff qn qnText existingBindings) (cfg ^. #topicBindings)--  pure (queueAction ++ notifyAction ++ fifoAction ++ bindingActions)---- | Reconcile a single topic binding.-reconcileBindingEff ::-  (Eff.Pgmq :> es) =>-  QueueName ->-  T.Text ->-  Set.Set (T.Text, T.Text) ->-  TopicPattern ->-  Eff es [ReconcileAction]-reconcileBindingEff qn qnText existingBindings pat =-  let patText = topicPatternToText pat-   in if Set.member (qnText, patText) existingBindings-        then pure [SkippedTopicBinding qn pat]-        else do-          Eff.bindTopic-            StmtTypes.BindTopic-              { topicPattern = pat,-                queueName = qn-              }-          pure [BoundTopic qn pat]+ensureQueuesReportEff = ensureQueuesReportWith effectfulOps
+ src/Pgmq/Config/Reconcile.hs view
@@ -0,0 +1,217 @@+-- | The backend-agnostic reconciliation core shared by "Pgmq.Config" (which+-- runs it in 'Hasql.Session.Session') and "Pgmq.Config.Effectful" (which runs+-- it in the @Pgmq@ effect). The logic lives here exactly once; the two public+-- modules only supply a t'ReconcileOps' record wiring the database calls.+--+-- This module is internal: it is listed under @other-modules@ in+-- @pgmq-config.cabal@ and is not part of the package's public API.+module Pgmq.Config.Reconcile+  ( ReconcileOps (..),+    ensureQueuesReportWith,+  )+where++import Control.Lens ((^.))+import Data.Generics.Labels ()+import Data.Int (Int32)+import Data.Map.Strict qualified as Map+import Data.Maybe (fromMaybe)+import Data.Set qualified as Set+import Data.Text qualified as T+import GHC.Generics (Generic)+import Pgmq.Config.Types+import Pgmq.Hasql.Statements.Types qualified as StmtTypes+import Pgmq.Types+  ( NotifyInsertThrottle,+    QueueName,+    TopicBinding,+    TopicPattern,+    UnvalidatedQueue,+    queueNameToText,+    topicPatternToText,+  )++-- | The database operations the reconciler needs, abstracted over the carrier+-- monad so one implementation serves both the 'Hasql.Session.Session' and+-- @Pgmq@-effect entry points.+data ReconcileOps m = ReconcileOps+  { -- | Deliberately the /unvalidated/ listing: the reconciler only compares+    -- observed names against declared ones, and re-validating them would make+    -- one foreign queue whose name 'Pgmq.Types.parseQueueName' rejects fail the+    -- whole reconcile at application startup.+    listQueuesUnvalidated :: m [UnvalidatedQueue],+    listTopicBindings :: m [TopicBinding],+    listNotifyInsertThrottles :: m [NotifyInsertThrottle],+    createQueue :: QueueName -> m (),+    createUnloggedQueue :: QueueName -> m (),+    createPartitionedQueue :: StmtTypes.CreatePartitionedQueue -> m (),+    enableNotifyInsert :: StmtTypes.EnableNotifyInsert -> m (),+    createFifoIndex :: QueueName -> m (),+    bindTopic :: StmtTypes.BindTopic -> m (),+    -- | Queues that already carry their FIFO headers index, read from the+    -- @pg_indexes@ catalog. pgmq has no index-existence function, so without+    -- this the reconciler cannot say truthfully whether it created one.+    listFifoIndexQueueNames :: m [T.Text],+    -- | The reconciler's only mutation of already-existing state: bring a+    -- throttle row's interval in line with the declared one.+    updateNotifyInsert :: StmtTypes.UpdateNotifyInsert -> m ()+  }+  deriving stock (Generic)++-- | Reconcile the declared configs against existing state, returning a report+-- of actions taken. Queries existing state first and skips operations that are+-- already satisfied.+ensureQueuesReportWith ::+  (Monad m) =>+  ReconcileOps m ->+  [QueueConfig] ->+  m [ReconcileAction]+ensureQueuesReportWith ops configs = do+  existingQueues <- ops ^. #listQueuesUnvalidated+  existingBindings <- ops ^. #listTopicBindings+  existingThrottles <- ops ^. #listNotifyInsertThrottles+  existingFifoIndexes <- ops ^. #listFifoIndexQueueNames++  let existingQueuesByName =+        Map.fromList [(q ^. #unvalidatedName, q) | q <- existingQueues]+      existingBindingSet =+        Set.fromList+          [ (b ^. #bindingQueueName, topicPatternToText (b ^. #bindingPattern))+          | b <- existingBindings+          ]+      -- The interval is kept, not just the name: dropping it is what made+      -- declared-versus-stored throttle drift invisible.+      existingNotifyByName =+        Map.fromList+          [(t ^. #throttleQueueName, t ^. #throttleIntervalMs) | t <- existingThrottles]+      -- The catalog reports the lowercased physical form pgmq derives table+      -- names from. Declared names are lowercase-only (parseQueueName rejects+      -- anything else), so matching them textually is exact.+      existingFifoSet = Set.fromList existingFifoIndexes++  concat+    <$> traverse+      (reconcileQueue ops existingQueuesByName existingBindingSet existingNotifyByName existingFifoSet)+      configs++-- | Reconcile a single queue config against existing state, returning actions taken.+reconcileQueue ::+  (Monad m) =>+  ReconcileOps m ->+  Map.Map T.Text UnvalidatedQueue ->+  Set.Set (T.Text, T.Text) ->+  Map.Map T.Text Int32 ->+  Set.Set T.Text ->+  QueueConfig ->+  m [ReconcileAction]+reconcileQueue ops existingQueues existingBindings existingNotify existingFifo cfg = do+  let qn = cfg ^. #queueName+      qnText = queueNameToText qn+      declaredType = cfg ^. #queueType++  -- Queue creation, or — when the queue is already there — a shape comparison.+  -- Drift is reported and never repaired: the only way to change a queue's type+  -- is to drop and recreate it, which would destroy its messages.+  queueAction <-+    case Map.lookup qnText existingQueues of+      Just observed ->+        let observedType = observedQueueType observed+         in pure+              [ if declaredShape declaredType == observedType+                  then SkippedQueue qn+                  else DetectedQueueTypeDrift qn declaredType observedType+              ]+      Nothing -> do+        case declaredType of+          StandardQueue ->+            (ops ^. #createQueue) qn+          UnloggedQueue ->+            (ops ^. #createUnloggedQueue) qn+          PartitionedQueue pc ->+            (ops ^. #createPartitionedQueue)+              StmtTypes.CreatePartitionedQueue+                { queueName = qn,+                  partitionInterval = pc ^. #partitionInterval,+                  retentionInterval = pc ^. #retentionInterval+                }+        pure [CreatedQueue qn declaredType]++  -- Notification. A missing row is enabled; a row whose interval already+  -- matches is left strictly alone (re-enabling would reset last_notified_at);+  -- a row whose interval differs is updated in place.+  notifyAction <- case cfg ^. #notifyInsert of+    Nothing -> pure []+    Just nc ->+      let declaredMs = fromMaybe defaultThrottleMs (nc ^. #throttleMs)+       in case Map.lookup qnText existingNotify of+            Nothing -> do+              (ops ^. #enableNotifyInsert)+                StmtTypes.EnableNotifyInsert+                  { queueName = qn,+                    throttleIntervalMs = nc ^. #throttleMs+                  }+              pure [EnabledNotify qn (nc ^. #throttleMs)]+            Just observedMs+              | observedMs == declaredMs -> pure [SkippedNotify qn]+              | otherwise -> do+                  (ops ^. #updateNotifyInsert)+                    StmtTypes.UpdateNotifyInsert+                      { queueName = qn,+                        throttleIntervalMs = declaredMs+                      }+                  pure [UpdatedNotifyThrottle qn observedMs declaredMs]++  -- FIFO index. The catalog snapshot makes the report truthful; the underlying+  -- pgmq.create_fifo_index is CREATE INDEX IF NOT EXISTS, so losing a race with+  -- a concurrent replica degrades to a no-op rather than an error.+  fifoAction <-+    if cfg ^. #fifoIndex+      then+        if Set.member qnText existingFifo+          then pure [SkippedFifoIndex qn]+          else do+            (ops ^. #createFifoIndex) qn+            pure [CreatedFifoIndex qn]+      else pure []++  -- Topic bindings+  bindingActions <- concat <$> traverse (reconcileBinding ops qn qnText existingBindings) (cfg ^. #topicBindings)++  pure (queueAction ++ notifyAction ++ fifoAction ++ bindingActions)++-- | The shape a declared config asks for, reduced to what @pgmq.list_queues()@+-- can actually report. Partition interval and retention are deliberately+-- dropped here: the listing does not expose them, so they are not drift-checked.+declaredShape :: QueueType -> ObservedQueueType+declaredShape StandardQueue = ObservedStandard+declaredShape UnloggedQueue = ObservedUnlogged+declaredShape (PartitionedQueue _) = ObservedPartitioned++-- | The shape an observed queue actually has, from the two booleans+-- @pgmq.list_queues()@ reports.+observedQueueType :: UnvalidatedQueue -> ObservedQueueType+observedQueueType q+  | q ^. #unvalidatedIsPartitioned = ObservedPartitioned+  | q ^. #unvalidatedIsUnlogged = ObservedUnlogged+  | otherwise = ObservedStandard++-- | Reconcile a single topic binding.+reconcileBinding ::+  (Monad m) =>+  ReconcileOps m ->+  QueueName ->+  T.Text ->+  Set.Set (T.Text, T.Text) ->+  TopicPattern ->+  m [ReconcileAction]+reconcileBinding ops qn qnText existingBindings pat =+  let patText = topicPatternToText pat+   in if Set.member (qnText, patText) existingBindings+        then pure [SkippedTopicBinding qn pat]+        else do+          (ops ^. #bindTopic)+            StmtTypes.BindTopic+              { topicPattern = pat,+                queueName = qn+              }+          pure [BoundTopic qn pat]
src/Pgmq/Config/Types.hs view
@@ -1,3 +1,8 @@+-- | The declarative vocabulary: what a queue topology looks like as Haskell+-- values, and what a reconciliation run reports back.+--+-- t'QueueConfig' is the declaration; 'ReconcileAction' is the report. Everything+-- here is re-exported from "Pgmq.Config", which is the module to import. module Pgmq.Config.Types   ( -- * Queue Configuration     QueueConfig (..),@@ -17,6 +22,10 @@      -- * Reconciliation Report     ReconcileAction (..),+    ObservedQueueType (..),++    -- * Defaults+    defaultThrottleMs,   ) where @@ -56,22 +65,83 @@  -- | Configuration for insert notifications (LISTEN/NOTIFY). data NotifyConfig = NotifyConfig-  { -- | Minimum milliseconds between notifications. Nothing uses pgmq default (250ms).+  { -- | Minimum milliseconds between notifications. Nothing uses the documented+    -- pgmq default (250 ms), applied via COALESCE in the pgmq-hasql statement so+    -- SQL NULL never reaches the function.     throttleMs :: !(Maybe Int32)   }   deriving stock (Generic, Show) +-- | The queue shape actually observed in the database.+--+-- @pgmq.list_queues()@ reports two booleans per queue, partitioned and+-- unlogged, which describe exactly these three states. It reports nothing about+-- a partitioned queue's interval or retention settings, so those are not+-- drift-checked — see 'DetectedQueueTypeDrift'.+data ObservedQueueType+  = ObservedStandard+  | ObservedUnlogged+  | ObservedPartitioned+  deriving stock (Eq, Show)+ -- | An action taken (or skipped) during queue reconciliation. data ReconcileAction-  = CreatedQueue !QueueName !QueueType-  | EnabledNotify !QueueName !(Maybe Int32)-  | CreatedFifoIndex !QueueName-  | BoundTopic !QueueName !TopicPattern-  | SkippedQueue !QueueName-  | SkippedNotify !QueueName-  | SkippedFifoIndex !QueueName-  | SkippedTopicBinding !QueueName !TopicPattern+  = -- | The queue did not exist and was created with the declared type.+    CreatedQueue !QueueName !QueueType+  | -- | No throttle row existed, so insert notifications were enabled with the+    -- declared interval ('Nothing' meaning 'defaultThrottleMs').+    EnabledNotify !QueueName !(Maybe Int32)+  | -- | The FIFO headers index did not exist and was created.+    CreatedFifoIndex !QueueName+  | -- | The topic binding did not exist and was created.+    BoundTopic !QueueName !TopicPattern+  | -- | A queue with this name already existed and its observed type matches+    -- what was declared. Nothing was issued.+    SkippedQueue !QueueName+  | -- | A throttle row already existed with the declared interval. Nothing was+    -- issued — in particular the row was /not/ re-enabled, which would reset+    -- its @last_notified_at@.+    SkippedNotify !QueueName+  | -- | The FIFO headers index already existed; nothing was issued.+    SkippedFifoIndex !QueueName+  | -- | The topic binding already existed; nothing was issued.+    SkippedTopicBinding !QueueName !TopicPattern+  | -- | The declared throttle interval differed from the database row, so the+    -- row was updated in place via @pgmq.update_notify_insert@. Fields: queue,+    -- observed interval, declared interval (in milliseconds).+    --+    -- This is the reconciler's only mutation of already-existing state. The+    -- update also resets the throttle's @last_notified_at@ to the epoch, so the+    -- next insert on that queue notifies immediately; that is a property of+    -- @pgmq.update_notify_insert@ itself, and it happens at most once per real+    -- configuration change.+    UpdatedNotifyThrottle !QueueName !Int32 !Int32+  | -- | The queue exists but its observed shape contradicts the declared one.+    -- Fields: queue, declared type, observed type.+    --+    -- Nothing was mutated and nothing will be: converting a queue between+    -- standard, unlogged, and partitioned means dropping and recreating it,+    -- destroying every message it holds, which a startup reconciler must never+    -- do. Resolving the drift is an operator decision. This action replaces+    -- 'SkippedQueue' for the queue it concerns, so the report still carries+    -- exactly one queue-existence action per declared config.+    --+    -- Only the three-way shape is compared. A declared 'PartitionedQueue'+    -- against an observed partitioned queue matches regardless of its interval+    -- and retention settings, because @pgmq.list_queues()@ does not report+    -- them.+    DetectedQueueTypeDrift !QueueName !QueueType !ObservedQueueType   deriving stock (Show)++-- | The throttle interval pgmq applies when none is given: 250 milliseconds.+--+-- A t'NotifyConfig' whose @throttleMs@ is 'Nothing' means \"use this value\".+-- The pgmq-hasql enable statement supplies it with a SQL @coalesce($2, 250)@,+-- and @pgmq.enable_notify_insert@ declares the same figure as its parameter+-- default, so a 'Nothing' config and a stored 250 agree and reconciliation does+-- not flap between them.+defaultThrottleMs :: Int32+defaultThrottleMs = 250  -- | Create a standard queue configuration with no extras. standardQueue :: QueueName -> QueueConfig
test/ConfigSpec.hs view
@@ -7,6 +7,7 @@  import Control.Lens ((^.)) import Data.Generics.Labels ()+import Data.Int (Int32) import Data.Text qualified as T import Data.Time (UTCTime) import Data.Word (Word32)@@ -31,7 +32,11 @@       testEnsureQueuesIdempotent pool,       testEnsureQueuesIncremental pool,       testEnsureQueuesWithNotify pool,+      testEnsureQueuesWithNotifyDefault pool,       testEnsureQueuesWithFifo pool,+      testEnsureQueuesSkipsExistingFifo pool,+      testEnsureQueuesUpdatesDriftedThrottle pool,+      testEnsureQueuesReportsQueueTypeDrift pool,       testEnsureQueuesWithTopicBinding pool,       testEnsureQueuesIsTrulyIdempotent pool,       testEnsureQueuesSilentIdempotent pool,@@ -133,6 +138,36 @@     any isSkippedNotify actions2   cleanupQueue pool qn +-- | A queue configured with @withNotifyInsert Nothing@ must reconcile cleanly,+-- twice.+--+-- @Nothing@ is documented as "use the pgmq default (250 ms)". Before the fix+-- the pgmq-hasql statement bound an SQL NULL for the throttle interval, which+-- @pgmq.enable_notify_insert@ inserted into a @NOT NULL@ column — a column+-- DEFAULT does not apply to an explicitly supplied NULL — so the call always+-- raised SQLSTATE 23502. Reconciliation is not one transaction: each statement+-- autocommits, so the queue creation stuck while the notify enable failed,+-- and every subsequent startup failed the same way forever.+testEnsureQueuesWithNotifyDefault :: Pool.Pool -> TestTree+testEnsureQueuesWithNotifyDefault pool = testCase "enables notify insert with the default throttle" $ do+  qn <- genQueueName+  let configs = [withNotifyInsert Nothing (standardQueue qn)]+  actions <- runSession pool (ensureQueuesReport configs)+  assertBool "first run should have EnabledNotify action" $+    any isEnabledNotify actions+  -- Second run must be a clean skip, proving the first run actually recorded+  -- a throttle row rather than failing.+  actions2 <- runSession pool (ensureQueuesReport configs)+  assertBool "second run should skip notify" $+    any isSkippedNotify actions2+  -- The recorded interval must be the documented 250ms default.+  throttles <- runSession pool Sessions.listNotifyInsertThrottles+  let mine = filter (\t -> (t ^. #throttleQueueName) == queueNameToText qn) throttles+  case mine of+    [t] -> (t ^. #throttleIntervalMs) @?= 250+    _ -> assertFailure $ "expected exactly one throttle row, got " <> show (length mine)+  cleanupQueue pool qn+ testEnsureQueuesWithFifo :: Pool.Pool -> TestTree testEnsureQueuesWithFifo pool = testCase "creates FIFO index" $ do   qn <- genQueueName@@ -142,6 +177,85 @@     any isFifoIndex actions   cleanupQueue pool qn +-- | The FIFO action must report what actually happened.+--+-- @pgmq.create_fifo_index@ is @CREATE INDEX IF NOT EXISTS@ and tells the caller+-- nothing, so the reconciler used to report 'CreatedFifoIndex' on every run+-- forever and 'SkippedFifoIndex' was unreachable dead code. The reconciler now+-- snapshots @pg_indexes@ first, so the second run skips the call outright.+testEnsureQueuesSkipsExistingFifo :: Pool.Pool -> TestTree+testEnsureQueuesSkipsExistingFifo pool = testCase "second run skips the existing FIFO index" $ do+  qn <- genQueueName+  let configs = [withFifoIndex (standardQueue qn)]+  actions1 <- runSession pool (ensureQueuesReport configs)+  assertBool+    ("first run should create the FIFO index, got " <> show actions1)+    (any isFifoIndex actions1)+  actions2 <- runSession pool (ensureQueuesReport configs)+  assertBool+    ("second run should skip the FIFO index, got " <> show actions2)+    (any isSkippedFifoIndex actions2)+  assertBool+    ("second run must not claim to have created it, got " <> show actions2)+    (not (any isFifoIndex actions2))+  cleanupQueue pool qn++-- | A declared throttle interval that differs from the stored one is applied.+--+-- This is the reconciler's single mutation of already-existing state. Before+-- the fix the declared value was compared only for presence, so changing it in+-- the config had no effect on the database and the report said 'SkippedNotify'.+testEnsureQueuesUpdatesDriftedThrottle :: Pool.Pool -> TestTree+testEnsureQueuesUpdatesDriftedThrottle pool = testCase "updates a drifted notify throttle" $ do+  qn <- genQueueName+  _ <- runSession pool (ensureQueuesReport [withNotifyInsert (Just 250) (standardQueue qn)])+  -- Redeclare with a different interval: the row must be brought in line.+  actions <- runSession pool (ensureQueuesReport [withNotifyInsert (Just 500) (standardQueue qn)])+  assertBool+    ("expected UpdatedNotifyThrottle " <> show qn <> " 250 500, got " <> show actions)+    (any (isUpdatedThrottle qn 250 500) actions)+  throttleFor pool qn >>= (@?= 500)+  -- A third run at the now-current value is a clean skip, so the update+  -- converges instead of firing on every startup.+  actions3 <- runSession pool (ensureQueuesReport [withNotifyInsert (Just 500) (standardQueue qn)])+  assertBool+    ("third run should skip notify, got " <> show actions3)+    (any isSkippedNotify actions3)+  cleanupQueue pool qn++-- | A declared queue type contradicting the live queue is reported, not fixed.+--+-- Converting a queue's type means dropping and recreating it, destroying every+-- message it holds, so the reconciler surfaces the contradiction and leaves the+-- decision to an operator. Before the fix it silently reported 'SkippedQueue'.+testEnsureQueuesReportsQueueTypeDrift :: Pool.Pool -> TestTree+testEnsureQueuesReportsQueueTypeDrift pool = testCase "reports queue-type drift without mutating" $ do+  qn <- genQueueName+  _ <- runSession pool (ensureQueuesReport [standardQueue qn])+  actions <- runSession pool (ensureQueuesReport [unloggedQueue qn])+  assertBool+    ("expected DetectedQueueTypeDrift for " <> show qn <> ", got " <> show actions)+    (any (isQueueTypeDrift qn ObservedStandard) actions)+  assertBool+    ("drift must replace the plain skip, got " <> show actions)+    (not (any isSkippedQueue actions))+  -- The queue is untouched: still standard, still there.+  queues <- runSession pool Sessions.listQueues+  case filter (\q -> (q ^. #name) == qn) queues of+    [q] -> do+      (q ^. #isUnlogged) @?= False+      (q ^. #isPartitioned) @?= False+    other -> assertFailure $ "expected exactly one queue row, got " <> show (length other)+  cleanupQueue pool qn++-- | The stored throttle interval for a queue.+throttleFor :: Pool.Pool -> QueueName -> IO Int32+throttleFor pool qn = do+  throttles <- runSession pool Sessions.listNotifyInsertThrottles+  case filter (\t -> (t ^. #throttleQueueName) == queueNameToText qn) throttles of+    [t] -> pure (t ^. #throttleIntervalMs)+    other -> assertFailure $ "expected exactly one throttle row, got " <> show (length other)+ testEnsureQueuesWithTopicBinding :: Pool.Pool -> TestTree testEnsureQueuesWithTopicBinding pool = testCase "binds topic pattern" $ do   qn <- genQueueName@@ -196,6 +310,23 @@ isFifoIndex (CreatedFifoIndex _) = True isFifoIndex _ = False +isSkippedFifoIndex :: ReconcileAction -> Bool+isSkippedFifoIndex (SkippedFifoIndex _) = True+isSkippedFifoIndex _ = False++isSkippedQueue :: ReconcileAction -> Bool+isSkippedQueue (SkippedQueue _) = True+isSkippedQueue _ = False++isUpdatedThrottle :: QueueName -> Int32 -> Int32 -> ReconcileAction -> Bool+isUpdatedThrottle qn observed declared (UpdatedNotifyThrottle q o d) =+  q == qn && o == observed && d == declared+isUpdatedThrottle _ _ _ _ = False++isQueueTypeDrift :: QueueName -> ObservedQueueType -> ReconcileAction -> Bool+isQueueTypeDrift qn observed (DetectedQueueTypeDrift q _ o) = q == qn && o == observed+isQueueTypeDrift _ _ _ = False+ isBoundTopic :: ReconcileAction -> Bool isBoundTopic (BoundTopic _ _) = True isBoundTopic _ = False@@ -209,6 +340,8 @@ actionForQueue qn (SkippedFifoIndex q) = q == qn actionForQueue qn (BoundTopic q _) = q == qn actionForQueue qn (SkippedTopicBinding q _) = q == qn+actionForQueue qn (UpdatedNotifyThrottle q _ _) = q == qn+actionForQueue qn (DetectedQueueTypeDrift q _ _) = q == qn  -- | Silent-variant version of 'testEnsureQueuesIdempotent': call 'ensureQueues' -- twice with the same standard-queue config and confirm the queue exists exactly
+ test/ForeignQueueSpec.hs view
@@ -0,0 +1,268 @@+{-# LANGUAGE CPP #-}+{-# LANGUAGE OverloadedStrings #-}++-- | A queue created by any other client must not break this application's boot.+--+-- pgmq queues are ordinary SQL tables, and the server's only queue-name check is+-- length (@pgmq.validate_queue_name@ rejects names over 47 characters and+-- nothing else). The Haskell validator 'Pgmq.Types.parseQueueName' is far+-- stricter, and the typed listing decoder re-validates every name read back from+-- the database — so one foreign queue named @billing-events@ used to make+-- @listQueues@ fail to decode, and with it the pgmq-config reconciler, whose+-- first step is that listing. Someone else's queue became your boot failure.+--+-- The reconciler now snapshots existing queues through the /unvalidated/ listing+-- and compares names as plain text. This module proves it: it seeds a+-- hyphen-named queue, shows the typed listing still rejects it while the+-- unvalidated listing reads it, and then reconciles a normal declared queue+-- successfully through both the Session and the effect backend.+--+-- The hyphen matters. A hyphenated name is rejected by every generation of+-- 'Pgmq.Types.parseQueueName' — the older @[A-Za-z0-9_]@ form and the current+-- lowercase-only one alike — so these assertions do not depend on which+-- validation rules are in force.+--+-- This module runs on its own dedicated PostgreSQL instance, never the+-- suite-shared pool: tasty runs specs concurrently, and a foreign row in the+-- shared database would make every concurrent typed-@listQueues@ call fail+-- (@ConfigSpec@ makes four of them).+module ForeignQueueSpec (tests) where++import Control.Exception (bracket)+import Control.Lens ((^.))+import Data.Generics.Labels ()+import Data.Int (Int64)+import Data.List.NonEmpty (NonEmpty (..))+import Data.Text (Text)+import Data.Text qualified as T+import Data.Word (Word32)+import Database.PostgreSQL.Migrate+  ( defaultRunOptions,+    migrationPlan,+    runMigrationPlan,+  )+import EphemeralPg qualified as Pg+import Hasql.Decoders qualified as D+import Hasql.Encoders qualified as E+import Hasql.Pool qualified as Pool+import Hasql.Pool.Config qualified as PoolConfig+import Hasql.Session (Session)+import Hasql.Session qualified as Session+import Hasql.Statement (Statement, preparable)+import Pgmq.Config (ReconcileAction (..), ensureQueuesReport, standardQueue)+import Pgmq.Hasql.Sessions qualified as Sessions+import Pgmq.Migration qualified as Migration+import Pgmq.Types (QueueName, UnvalidatedQueue, parseQueueName)+import System.Random (randomRIO)+import Test.Tasty (TestTree, testGroup, withResource)+import Test.Tasty.HUnit (assertBool, assertFailure, testCase, (@?=))+#ifdef PGMQ_EFFECTFUL+import Effectful (runEff)+import Effectful.Error.Static (runError)+import Pgmq.Config.Effectful (ensureQueuesReportEff)+import Pgmq.Effectful.Interpreter (PgmqRuntimeError, runPgmq)+#endif++-- | The foreign queue name. Hyphens are legal server-side and rejected by every+-- generation of 'Pgmq.Types.parseQueueName'.+foreignName :: Text+foreignName = "billing-events"++-- | Everything one pass against the seeded database observed, so each fact can+-- be asserted as its own test case.+data ForeignQueueObservations = ForeignQueueObservations+  { -- | The queue declared through the Session backend.+    obsQueueName :: !QueueName,+    -- | The queue declared through the effect backend.+    obsEffQueueName :: !QueueName,+    -- | The error the typed listing failed with, if it failed at all.+    obsTypedListError :: !(Maybe String),+    -- | The foreign row as seen by the unvalidated listing.+    obsUnvalidatedForeign :: !(Maybe UnvalidatedQueue),+    -- | Report of the first Session-backed reconcile.+    obsFirstReport :: ![ReconcileAction],+    -- | Report of an immediately repeated reconcile.+    obsSecondReport :: ![ReconcileAction],+    -- | @pgmq.meta@ rows still naming the foreign queue afterwards.+    obsForeignMetaCount :: !Int64,+    -- | Report of the effect-backed reconcile.+    obsEffReport :: ![ReconcileAction]+  }++tests :: TestTree+tests =+  withResource runForeignQueueCycle (const (pure ())) $ \getObs ->+    testGroup+      "ForeignQueueSpec"+      ( [ testCase "typed listQueues rejects a foreign name (evidence)" $ do+            obs <- getObs+            case obsTypedListError obs of+              Nothing ->+                assertFailure $+                  "expected the typed listQueues to fail decoding "+                    <> show foreignName+                    <> ", but it succeeded"+              Just _ -> pure ()+            case obsUnvalidatedForeign obs of+              Nothing ->+                assertFailure $+                  "the unvalidated listing did not return a row named " <> show foreignName+              Just q -> do+                (q ^. #unvalidatedName) @?= foreignName+                (q ^. #unvalidatedIsPartitioned) @?= False+                (q ^. #unvalidatedIsUnlogged) @?= False,+          testCase "ensureQueues succeeds despite a foreign queue" $ do+            obs <- getObs+            assertBool+              ( "expected CreatedQueue "+                  <> show (obsQueueName obs)+                  <> " in the first report, got "+                  <> show (obsFirstReport obs)+              )+              (any (isCreatedQueue (obsQueueName obs)) (obsFirstReport obs))+            assertBool+              ( "expected SkippedQueue "+                  <> show (obsQueueName obs)+                  <> " in the second report, got "+                  <> show (obsSecondReport obs)+              )+              (any (isSkippedQueue (obsQueueName obs)) (obsSecondReport obs))+            obsForeignMetaCount obs @?= 1+        ]+          <> effectfulCases getObs+      )++-- | The effect-backed parity case, present only when the library's @effectful@+-- flag is on — with it off, "Pgmq.Config.Effectful" is not built at all.+effectfulCases :: IO ForeignQueueObservations -> [TestTree]+#ifdef PGMQ_EFFECTFUL+effectfulCases getObs =+  [ testCase "effectful ensureQueues matches" $ do+      obs <- getObs+      assertBool+        ( "expected CreatedQueue "+            <> show (obsEffQueueName obs)+            <> " in the effectful report, got "+            <> show (obsEffReport obs)+        )+        (any (isCreatedQueue (obsEffQueueName obs)) (obsEffReport obs))+  ]+#else+effectfulCases _ = []+#endif++isCreatedQueue :: QueueName -> ReconcileAction -> Bool+isCreatedQueue qn (CreatedQueue q _) = q == qn+isCreatedQueue _ _ = False++isSkippedQueue :: QueueName -> ReconcileAction -> Bool+isSkippedQueue qn (SkippedQueue q) = q == qn+isSkippedQueue _ _ = False++-- | Start a dedicated PostgreSQL instance, seed the foreign queue, reconcile+-- through both backends, and record what happened. The instance is stopped+-- before this returns.+runForeignQueueCycle :: IO ForeignQueueObservations+runForeignQueueCycle =+  bracket startOrFail Pg.stop $ \db -> do+    installPgmq db+    bracket (acquirePool db) Pool.release observe++observe :: Pool.Pool -> IO ForeignQueueObservations+observe pool = do+  qn <- genQueueName "foreign_test_"+  effQn <- genQueueName "foreign_eff_"++  -- Seed a queue no Haskell client could have created. `%I` quoting inside+  -- pgmq.create means the hyphen reaches the physical table name intact.+  runSession pool (Session.script "select pgmq.create('billing-events')")++  typedResult <- Pool.use pool Sessions.listQueues+  unvalidated <- runSession pool Sessions.listQueuesUnvalidated++  firstReport <- runSession pool (ensureQueuesReport [standardQueue qn])+  secondReport <- runSession pool (ensureQueuesReport [standardQueue qn])+  metaCount <- runSession pool (Session.statement foreignName metaRowCount)++  effReport <- runEffectfulReconcile pool effQn++  pure+    ForeignQueueObservations+      { obsQueueName = qn,+        obsEffQueueName = effQn,+        obsTypedListError = either (Just . show) (const Nothing) typedResult,+        obsUnvalidatedForeign =+          case filter (\q -> q ^. #unvalidatedName == foreignName) unvalidated of+            (q : _) -> Just q+            [] -> Nothing,+        obsFirstReport = firstReport,+        obsSecondReport = secondReport,+        obsForeignMetaCount = metaCount,+        obsEffReport = effReport+      }++-- | Reconcile through the @Pgmq@ effect with the plain interpreter, so the test+-- pins that /both/ backends read the lenient listing. With the @effectful@ flag+-- off this returns no actions and 'effectfulCases' emits no test case.+runEffectfulReconcile :: Pool.Pool -> QueueName -> IO [ReconcileAction]+#ifdef PGMQ_EFFECTFUL+runEffectfulReconcile pool qn = do+  result <-+    runEff . runError @PgmqRuntimeError . runPgmq pool $+      ensureQueuesReportEff [standardQueue qn]+  case result of+    Left (_cs, err) ->+      assertFailure $ "effect-backed ensureQueuesReportEff failed: " <> show err+    Right actions -> pure actions+#else+runEffectfulReconcile _pool _qn = pure []+#endif++-- Database plumbing -----------------------------------------------------------++startOrFail :: IO Pg.Database+startOrFail = do+  result <- Pg.startCached Pg.defaultConfig Pg.defaultCacheConfig+  case result of+    Left err -> assertFailure $ "could not start a dedicated PostgreSQL: " <> show err+    Right db -> pure db++-- | Apply the full pgmq migration ledger, exactly as @EphemeralDb@ does.+installPgmq :: Pg.Database -> IO ()+installPgmq db = do+  component <- either (error . ("Invalid PGMQ migration component: " <>) . show) pure Migration.pgmqMigrations+  plan <- either (error . ("Invalid PGMQ migration plan: " <>) . show) pure (migrationPlan (component :| []))+  installResult <- runMigrationPlan defaultRunOptions (Pg.connectionSettings db) plan+  case installResult of+    Left migrationErr -> error $ "Migration failed: " <> show migrationErr+    Right _ -> pure ()++acquirePool :: Pg.Database -> IO Pool.Pool+acquirePool db =+  Pool.acquire $+    PoolConfig.settings+      [ PoolConfig.size 2,+        PoolConfig.staticConnectionSettings (Pg.connectionSettings db)+      ]++runSession :: Pool.Pool -> Session a -> IO a+runSession pool session = do+  result <- Pool.use pool session+  case result of+    Left err -> assertFailure $ "Session failed: " <> show err+    Right a -> pure a++-- | How many @pgmq.meta@ rows name the given queue.+metaRowCount :: Statement Text Int64+metaRowCount = preparable sql encoder decoder+  where+    sql = "select count(*)::int8 from pgmq.meta where queue_name = $1"+    encoder = E.param (E.nonNullable E.text)+    decoder = D.singleRow (D.column (D.nonNullable D.int8))++genQueueName :: Text -> IO QueueName+genQueueName prefix = do+  suffix <- randomRIO (10000 :: Word32, 99999)+  case parseQueueName (prefix <> T.pack (show suffix)) of+    Left err -> error $ "Failed to generate queue name: " <> show err+    Right qn -> pure qn
test/Main.hs view
@@ -4,6 +4,8 @@  import ConfigSpec qualified import EphemeralDb (withPgmqDb)+import ForeignQueueSpec qualified+import NotifyCrashSpec qualified import Test.Tasty (defaultMain, testGroup)  main :: IO ()@@ -12,7 +14,15 @@     let tree =           testGroup             "pgmq-config"-            [ ConfigSpec.tests pool+            [ ConfigSpec.tests pool,+              -- NotifyCrashSpec manages its own PostgreSQL instance: it crashes+              -- the server, which the shared pool above could not survive.+              NotifyCrashSpec.tests,+              -- ForeignQueueSpec seeds a queue whose name parseQueueName+              -- rejects; on the shared pool above that row would fail every+              -- concurrent typed listQueues call, so it too runs on its own+              -- PostgreSQL instance.+              ForeignQueueSpec.tests             ]     defaultMain tree   case result of
+ test/NotifyCrashSpec.hs view
@@ -0,0 +1,296 @@+{-# LANGUAGE OverloadedStrings #-}++-- | PGH-6: insert notifications must survive PostgreSQL crash recovery.+--+-- @pgmq.notify_insert_throttle@ is an UNLOGGED table, so PostgreSQL truncates it+-- during crash recovery. The insert trigger only raised @PG_NOTIFY@ when it+-- successfully updated a throttle row, so after a crash the trigger fired, found+-- no row, and silently stopped notifying until an application re-enabled notify.+--+-- This module drives a real crash cycle: it starts its own PostgreSQL instance+-- (never the suite-shared one), enables notify, kills the server with an+-- immediate shutdown (SIGQUIT), restarts it on the same data directory, and then+-- asserts that a post-recovery send still reaches a LISTENing client.+module NotifyCrashSpec (tests) where++import Control.Concurrent (threadDelay)+import Control.Exception (bracket)+import Control.Lens ((^.))+import Control.Monad (unless)+import Data.Aeson qualified as Aeson+import Data.ByteString (ByteString)+import Data.Generics.Labels ()+import Data.IORef (IORef, newIORef, readIORef, writeIORef)+import Data.Int (Int64)+import Data.List.NonEmpty (NonEmpty (..))+import Data.Text (Text)+import Data.Text qualified as T+import Data.Text.Encoding qualified as TE+import Data.Word (Word32)+import Database.PostgreSQL.LibPQ qualified as LibPQ+import Database.PostgreSQL.Migrate+  ( defaultRunOptions,+    migrationPlan,+    runMigrationPlan,+  )+import EphemeralPg qualified as Pg+-- 'shutdownMode' names a field of both Pg.Config and Pg.Database, so the record+-- update below needs the selector from the module that defines only Database.+import EphemeralPg.Database qualified as PgDb+import Hasql.Decoders qualified as D+import Hasql.Encoders qualified as E+import Hasql.Pool qualified as Pool+import Hasql.Pool.Config qualified as PoolConfig+import Hasql.Session (Session)+import Hasql.Session qualified as Session+import Hasql.Statement (Statement, preparable)+import Pgmq.Config (ensureQueues, standardQueue, withNotifyInsert)+import Pgmq.Hasql.Sessions qualified as Sessions+import Pgmq.Hasql.Statements.Types qualified as StmtTypes+import Pgmq.Migration qualified as Migration+import Pgmq.Types (MessageBody (..), QueueName, notifyChannelName, parseQueueName, queueNameToText)+import System.Random (randomRIO)+import Test.Tasty (TestTree, testGroup, withResource)+import Test.Tasty.HUnit (assertBool, assertFailure, testCase, (@?=))++-- | Everything the crash cycle observed, collected in one pass so the+-- assertions below can report each fact as its own test case.+data CrashObservations = CrashObservations+  { -- | The queue the cycle ran against.+    obsQueueName :: !QueueName,+    -- | The channel the listener subscribed to.+    obsChannel :: !Text,+    -- | Queue names present in @pgmq.list_notify_insert_throttles()@ after the crash.+    obsThrottlesAfterCrash :: ![Text],+    -- | Insert triggers on the queue table after the crash (1 = survived).+    obsTriggersAfterCrash :: !Int64,+    -- | Messages still in the queue after the crash (the queue table is logged).+    obsQueueLengthAfterCrash :: !Int64,+    -- | Channel of the notification delivered by the post-crash send, if any.+    obsNotifyAfterCrash :: !(Maybe ByteString),+    -- | Queue names in the throttle table after a reconcile ran.+    obsThrottlesAfterReconcile :: ![Text]+  }++tests :: TestTree+tests =+  withResource runCrashCycle (const (pure ())) $ \getObs ->+    testGroup+      "NotifyCrashSpec"+      [ testCase "post-crash: throttle row truncated, trigger intact" $ do+          obs <- getObs+          obsThrottlesAfterCrash obs @?= []+          obsTriggersAfterCrash obs @?= 1+          obsQueueLengthAfterCrash obs @?= 1,+        testCase "post-crash: send delivers a notification" $ do+          obs <- getObs+          case obsNotifyAfterCrash obs of+            Nothing ->+              assertFailure $+                "expected a notification on "+                  <> show (obsChannel obs)+                  <> " within 2s after crash recovery, got none"+            Just chan ->+              assertBool+                ( "notification arrived on "+                    <> show chan+                    <> " but the contract channel is "+                    <> show (obsChannel obs)+                )+                (chan == TE.encodeUtf8 (obsChannel obs)),+        testCase "post-crash: a reconcile restores the throttle row" $ do+          obs <- getObs+          obsThrottlesAfterReconcile obs @?= [queueNameToText (obsQueueName obs)]+      ]++-- | Start a dedicated PostgreSQL instance, enable notify, crash it, recover it,+-- and record what happened. Every resource is released before this returns.+runCrashCycle :: IO CrashObservations+runCrashCycle = do+  qn <- genQueueName+  db0 <- startOrFail+  ref <- newIORef db0+  bracket (pure ref) (\r -> readIORef r >>= Pg.stop) (crashCycle qn)++crashCycle :: QueueName -> IORef Pg.Database -> IO CrashObservations+crashCycle qn ref = do+  db0 <- readIORef ref+  installPgmq db0++  -- Pre-crash: create the queue, enable unthrottled notify, send one message.+  -- No listener is opened yet: an immediate shutdown kills every pre-crash+  -- connection, so a listener created here could never see the assertion's+  -- notification.+  bracket (acquirePool db0) Pool.release $ \pool -> do+    runSession pool (Sessions.createQueue qn)+    runSession pool $+      Sessions.enableNotifyInsert+        StmtTypes.EnableNotifyInsert+          { StmtTypes.queueName = qn,+            StmtTypes.throttleIntervalMs = Just 0+          }+    runSession pool (sendProbe qn "before-crash")+    -- ephemeral-pg runs PostgreSQL with fsync, synchronous_commit and+    -- full_page_writes all off, so an immediate shutdown would otherwise discard+    -- every commit still sitting in the WAL buffers — including the pgmq schema+    -- itself. CHECKPOINT flushes them. It does NOT make the unlogged throttle+    -- table crash-safe: recovery still resets unlogged relations to their init+    -- fork, which is the behavior under test.+    runSession pool (Session.script "checkpoint")++  db1 <- crashAndRecover db0+  writeIORef ref db1++  bracket (acquirePool db1) Pool.release $ \pool -> do+    throttlesAfterCrash <- listThrottleNames pool+    triggers <- runSession pool (Session.statement (queueTableName qn) insertTriggerCount)+    metrics <- runSession pool (Sessions.queueMetrics qn)++    let channel = notifyChannelName qn+    notified <- withListener db1 channel $ \conn -> do+      _ <- runSession pool (sendProbe qn "after-crash")+      awaitNotify conn 20++    runSession pool (ensureQueues [withNotifyInsert (Just 0) (standardQueue qn)])+    throttlesAfterReconcile <- listThrottleNames pool++    pure+      CrashObservations+        { obsQueueName = qn,+          obsChannel = channel,+          obsThrottlesAfterCrash = throttlesAfterCrash,+          obsTriggersAfterCrash = triggers,+          obsQueueLengthAfterCrash = metrics ^. #queueLength,+          obsNotifyAfterCrash = LibPQ.notifyRelname <$> notified,+          obsThrottlesAfterReconcile = throttlesAfterReconcile+        }++-- | Stop PostgreSQL with SIGQUIT and start it again on the same data directory.+-- That is a genuine crash: the next start runs crash recovery, which truncates+-- every UNLOGGED table. Retries once, then fails loudly — the crash cycle is the+-- test, so it must never degrade into a skip.+crashAndRecover :: Pg.Database -> IO Pg.Database+crashAndRecover db = do+  let crashing = db {PgDb.shutdownMode = Pg.ShutdownImmediate}+  first <- Pg.restart crashing+  case first of+    Right db' -> pure db'+    Left _ -> do+      second <- Pg.restart crashing+      case second of+        Right db' -> pure db'+        Left err -> assertFailure $ "could not restart PostgreSQL after crash: " <> show err++-- Database plumbing -----------------------------------------------------------++startOrFail :: IO Pg.Database+startOrFail = do+  result <- Pg.startCached Pg.defaultConfig Pg.defaultCacheConfig+  case result of+    Left err -> assertFailure $ "could not start a dedicated PostgreSQL: " <> show err+    Right db -> pure db++-- | Apply the full pgmq migration ledger, exactly as @EphemeralDb@ does.+installPgmq :: Pg.Database -> IO ()+installPgmq db = do+  component <- either (error . ("Invalid PGMQ migration component: " <>) . show) pure Migration.pgmqMigrations+  plan <- either (error . ("Invalid PGMQ migration plan: " <>) . show) pure (migrationPlan (component :| []))+  installResult <- runMigrationPlan defaultRunOptions (Pg.connectionSettings db) plan+  case installResult of+    Left migrationErr -> error $ "Migration failed: " <> show migrationErr+    Right _ -> pure ()++acquirePool :: Pg.Database -> IO Pool.Pool+acquirePool db =+  Pool.acquire $+    PoolConfig.settings+      [ PoolConfig.size 2,+        PoolConfig.staticConnectionSettings (Pg.connectionSettings db)+      ]++runSession :: Pool.Pool -> Session a -> IO a+runSession pool session = do+  result <- Pool.use pool session+  case result of+    Left err -> assertFailure $ "Session failed: " <> show err+    Right a -> pure a++listThrottleNames :: Pool.Pool -> IO [Text]+listThrottleNames pool = do+  throttles <- runSession pool Sessions.listNotifyInsertThrottles+  pure (map (^. #throttleQueueName) throttles)++sendProbe :: QueueName -> Text -> Session ()+sendProbe qn label =+  ()+    <$ Sessions.sendMessage+      StmtTypes.SendMessage+        { StmtTypes.queueName = qn,+          StmtTypes.messageBody = MessageBody (Aeson.String label),+          StmtTypes.delay = Nothing+        }++-- | The physical table backing a queue: @q_@ plus the lowercased queue name.+queueTableName :: QueueName -> Text+queueTableName qn = "q_" <> T.toLower (queueNameToText qn)++-- | How many insert-notification triggers exist on the given @pgmq@ table.+insertTriggerCount :: Statement Text Int64+insertTriggerCount = preparable sql encoder decoder+  where+    sql =+      "select count(*)::int8 \+      \from pg_trigger t \+      \join pg_class c on c.oid = t.tgrelid \+      \join pg_namespace n on n.oid = c.relnamespace \+      \where n.nspname = 'pgmq' \+      \and c.relname = $1 \+      \and t.tgname = 'trigger_notify_queue_insert_listeners'"+    encoder = E.param (E.nonNullable E.text)+    decoder = D.singleRow (D.column (D.nonNullable D.int8))++genQueueName :: IO QueueName+genQueueName = do+  suffix <- randomRIO (10000 :: Word32, 99999)+  case parseQueueName ("crash_test_" <> T.pack (show suffix)) of+    Left err -> error $ "Failed to generate queue name: " <> show err+    Right qn -> pure qn++-- LISTEN plumbing -------------------------------------------------------------++-- | Open a raw libpq connection (hasql exposes no notification API), subscribe+-- to @channel@, and run the action. @ephemeral-pg@ hands out connection strings+-- as 'Text' while libpq consumes 'ByteString', so both the conninfo and the+-- command are encoded explicitly.+withListener :: Pg.Database -> Text -> (LibPQ.Connection -> IO a) -> IO a+withListener db channel action =+  bracket (LibPQ.connectdb (TE.encodeUtf8 (Pg.connectionString db))) LibPQ.finish $ \conn -> do+    connStatus <- LibPQ.status conn+    unless (connStatus == LibPQ.ConnectionOk) $ do+      err <- LibPQ.errorMessage conn+      assertFailure $ "libpq connection failed: " <> show err+    -- The channel contains dots, so LISTEN needs the identifier double-quoted.+    result <- LibPQ.exec conn (TE.encodeUtf8 ("LISTEN " <> quoteIdentifier channel))+    case result of+      Nothing -> assertFailure "LISTEN returned no result"+      Just res -> do+        execStatus <- LibPQ.resultStatus res+        unless (execStatus == LibPQ.CommandOk) $+          assertFailure ("LISTEN failed with " <> show execStatus)+    action conn++quoteIdentifier :: Text -> Text+quoteIdentifier ident = "\"" <> T.replace "\"" "\"\"" ident <> "\""++-- | Poll for a notification, 100 ms per attempt.+awaitNotify :: LibPQ.Connection -> Int -> IO (Maybe LibPQ.Notify)+awaitNotify conn attempts+  | attempts <= 0 = pure Nothing+  | otherwise = do+      _ <- LibPQ.consumeInput conn+      pending <- LibPQ.notifies conn+      case pending of+        Just n -> pure (Just n)+        Nothing -> do+          threadDelay 100_000+          awaitNotify conn (attempts - 1)