diff --git a/CHANGELOG.md b/CHANGELOG.md --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -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. diff --git a/pgmq-config.cabal b/pgmq-config.cabal --- a/pgmq-config.cabal +++ b/pgmq-config.cabal @@ -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 diff --git a/src/Pgmq/Config.hs b/src/Pgmq/Config.hs --- a/src/Pgmq/Config.hs +++ b/src/Pgmq/Config.hs @@ -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 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 @@ -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 diff --git a/src/Pgmq/Config/Reconcile.hs b/src/Pgmq/Config/Reconcile.hs new file mode 100644 --- /dev/null +++ b/src/Pgmq/Config/Reconcile.hs @@ -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] diff --git a/src/Pgmq/Config/Types.hs b/src/Pgmq/Config/Types.hs --- a/src/Pgmq/Config/Types.hs +++ b/src/Pgmq/Config/Types.hs @@ -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 diff --git a/test/ConfigSpec.hs b/test/ConfigSpec.hs --- a/test/ConfigSpec.hs +++ b/test/ConfigSpec.hs @@ -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 diff --git a/test/ForeignQueueSpec.hs b/test/ForeignQueueSpec.hs new file mode 100644 --- /dev/null +++ b/test/ForeignQueueSpec.hs @@ -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 diff --git a/test/Main.hs b/test/Main.hs --- a/test/Main.hs +++ b/test/Main.hs @@ -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 diff --git a/test/NotifyCrashSpec.hs b/test/NotifyCrashSpec.hs new file mode 100644 --- /dev/null +++ b/test/NotifyCrashSpec.hs @@ -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)