pgmq-hasql 0.4.0.1 → 0.5.0.0

raw patch · 22 files changed

+1601/−50 lines, 22 filesdep +postgresql-libpqdep ~pgmq-core

Dependencies added: postgresql-libpq

Dependency ranges changed: pgmq-core

Files

• CHANGELOG.md +73/−0
• pgmq-hasql.cabal +18/−11
• src/Pgmq.hs +8/−0
• src/Pgmq/Hasql/Decoders.hs +32/−1
• src/Pgmq/Hasql/Encoders.hs +2/−1
• src/Pgmq/Hasql/Sessions.hs +20/−3
• src/Pgmq/Hasql/Statements/Message.hs +33/−12
• src/Pgmq/Hasql/Statements/QueueManagement.hs +13/−2
• src/Pgmq/Hasql/Statements/QueueObservability.hs +31/−2
• src/Pgmq/Hasql/Statements/Types.hs +17/−3
• test/AdvancedOpsSpec.hs +3/−2
• test/AliasingSpec.hs +228/−0
• test/AllFunctionsDecoderSpec.hs +4/−2
• test/EphemeralDb.hs +9/−5
• test/Main.hs +21/−2
• test/MessageSpec.hs +3/−2
• test/MixedCaseRemediationSpec.hs +358/−0
• test/NotifyChannelSpec.hs +129/−0
• test/NotifyRaceSpec.hs +103/−0
• test/NullBodySpec.hs +158/−0
• test/NullSemanticsSpec.hs +300/−0
• test/QueueSpec.hs +38/−2

@@ -1,5 +1,78 @@ # Revision history for pgmq-hasql +## 0.5.0.0 -- 2026-08-06++### Breaking Changes++* `changeVisibilityTimeout` and `setVisibilityTimeoutAt` now return `Maybe Message`+  instead of `Message`, at both the statement and session layers. `pgmq.set_vt` is+  `RETURNS SETOF` and yields zero rows when the target message no longer exists (already+  deleted, archived, or popped). Decoding that with a single-row decoder raised an+  `UnexpectedRowCountStatementError` — the same error shape a genuine infrastructure+  failure has — so a caller extending a lease could not distinguish a lost race from a+  broken database. Callers that used the result must now handle `Nothing`; callers that+  discarded it compile unchanged. The batch variants are unaffected.+* Queue names read back from the database are re-validated by `queueDecoder` against the+  tightened `parseQueueName` in pgmq-core 0.5. See that package's changelog for the+  required `pgmq.meta` remediation, and `listQueuesUnvalidated` below for the lenient+  read.++### New Features++* `notifyChannelName` is re-exported from the `Pgmq` umbrella module (defined in+  pgmq-core). Use it instead of assembling the LISTEN/NOTIFY channel name by hand — the+  name this package previously documented was wrong; see Documentation below.+* `listQueuesUnvalidated` (statement, session, and `Pgmq` re-export) reads `pgmq.meta`+  with the queue name decoded as `Text`, yielding `UnvalidatedQueue` rows. pgmq's+  server-side validator checks only length, so any co-tenant client can create a name+  `parseQueueName` rejects, and the typed `listQueues` decoder fails the whole listing on+  one such row. The typed `listQueues` keeps its strict decoding for API consumers.+* `listFifoIndexQueueNames` reports which queues already carry a `q_<name>_fifo_idx`. pgmq exposes+  no index-existence query — `create_fifo_index` delegates to `CREATE INDEX IF NOT EXISTS`+  and reports nothing back — so this reads the `pg_indexes` catalog view. It is the first+  statement in this package that queries a PostgreSQL catalog rather than calling a+  `pgmq.*` function.++### Bug Fixes++* `pop` with `qty = Nothing` now pops one message, as documented. It previously deleted+  and returned every visible message in the queue. The `Maybe` parameter was encoded as a+  nullable bind, so `Nothing` reached PostgreSQL as SQL NULL; a plpgsql parameter+  `DEFAULT` applies only to omitted arguments, and NULL in a `LIMIT` clause means+  `LIMIT ALL`. Because `pop` deletes, there was no visibility timeout to recover the+  messages.+* `readMessage` and `readWithPoll` with `batchSize = Nothing` now read one message, as+  documented. They previously leased the entire queue through the same `LIMIT NULL` path,+  hiding every message from other consumers for the visibility timeout.+* `enableNotifyInsert` with `throttleIntervalMs = Nothing` now installs the documented+  250 ms throttle. It previously failed with SQLSTATE 23502 on every call, because a+  column `DEFAULT` does not apply to an explicitly supplied NULL.+* `ReadMessage.conditional` now filters. The field existed and was documented, but was+  never encoded, so a `Just` filter was silently ignored and every visible message was+  returned. `readWithPoll`'s conditional already worked.+* A message whose body is SQL NULL no longer poisons every read batch. The `message`+  column is nullable and `pgmq.send('q', NULL::jsonb)` is legal SQL for any non-Haskell+  producer; one such row made every batch containing it fail at decode — after the read+  statement had already bumped `vt` and `read_ct` for the whole batch — and the row could+  not be seen or archived through this client. A SQL NULL body now decodes as JSON `null`+  (`MessageBody Aeson.Null`, deliberately indistinguishable from an explicitly-sent JSON+  `null` body), so the row is readable, identifiable, and archivable through the normal+  API.++### Documentation++* The documented LISTEN/NOTIFY channel name was wrong. `enableNotifyInsert` claimed+  notifications arrive on `pgmq_<queue_name>`; the real channel is+  `pgmq.q_<lowercased queue name>.INSERT`, so anyone following the documentation listened+  on a channel that never receives anything. Corrected on the Haddock and in+  `docs/design/006-queue-notifications.md`, and replaced by `notifyChannelName`. The full+  contract — including the poll-fallback requirement and the crash fail-open semantics —+  is in `docs/design/015-notification-delivery-contract.md`.++### Other Changes++* Bumped `pgmq-core` dependency bound to `>=0.5 && <0.6`.+ ## 0.4.0.1 -- 2026-07-14  * Version bump only — coordinated release with pgmq-migration 0.4.0.1.

@@ -1,6 +1,6 @@ cabal-version:   3.4 name:            pgmq-hasql-version:         0.4.0.1+version:         0.5.0.0 synopsis:        Hasql-based client for PGMQ (PostgreSQL Message Queue) description:   A Haskell client library for PGMQ (PostgreSQL Message Queue) built@@ -57,7 +57,7 @@     , hasql              ^>=1.10     , hasql-transaction  ^>=1.2     , lens               ^>=5.3-    , pgmq-core          >=0.4   && <0.5+    , pgmq-core          >=0.5   && <0.6     , template-haskell   >=2.20  && <3     , text               ^>=2.1     , time               ^>=1.14@@ -75,12 +75,18 @@   ghc-options:        -threaded -rtsopts -with-rtsopts=-N   other-modules:     AdvancedOpsSpec+    AliasingSpec     AllFunctionsDecoderSpec     DecoderValidationSpec     EphemeralDb     Generators     MessageSpec     MetricsSpec+    MixedCaseRemediationSpec+    NotifyChannelSpec+    NotifyRaceSpec+    NullBodySpec+    NullSemanticsSpec     QueueSpec     RoundTripSpec     SchemaSpec@@ -93,20 +99,21 @@    build-depends:     , aeson-    , base            >=4.18  && <5-    , ephemeral-pg    >=0.2.1+    , base              >=4.18   && <5+    , ephemeral-pg      >=0.2.1     , hasql-    , hasql-pool      ^>=1.4-    , hedgehog        ^>=1.5+    , hasql-pool        ^>=1.4+    , hedgehog          ^>=1.5     , pg-migrate     , pgmq-core     , pgmq-hasql     , pgmq-migration-    , random          ^>=1.2-    , scientific      ^>=0.3-    , tasty           ^>=1.5-    , tasty-hedgehog  ^>=1.4-    , tasty-hunit     ^>=0.10+    , postgresql-libpq  >=0.10.1 && <0.12+    , random            ^>=1.2+    , scientific        ^>=0.3+    , tasty             ^>=1.5+    , tasty-hedgehog    ^>=1.4+    , tasty-hunit       ^>=0.10     , text     , time     , vector

@@ -34,6 +34,8 @@     setVisibilityTimeoutAt,     batchSetVisibilityTimeoutAt,     listQueues,+    listQueuesUnvalidated,+    listFifoIndexQueueNames,     readWithPoll,     pop,     queueMetrics,@@ -68,6 +70,7 @@     MessageId (..),     Message (..),     Queue (..),+    UnvalidatedQueue (..),     QueueName,     SendMessage (..),     SendMessageForLater (..),@@ -105,6 +108,7 @@     RoutingMatch (..),     TopicSendResult (..),     NotifyInsertThrottle (..),+    notifyChannelName,     BindTopic (..),     UnbindTopic (..),     SendTopic (..),@@ -147,8 +151,10 @@     disableNotifyInsert,     dropQueue,     enableNotifyInsert,+    listFifoIndexQueueNames,     listNotifyInsertThrottles,     listQueues,+    listQueuesUnvalidated,     listTopicBindings,     listTopicBindingsForQueue,     pop,@@ -212,6 +218,8 @@     TopicBinding (..),     TopicPattern,     TopicSendResult (..),+    UnvalidatedQueue (..),+    notifyChannelName,     parseQueueName,     parseRoutingKey,     parseTopicPattern,

@@ -2,6 +2,7 @@   ( messageDecoder,     messageIdDecoder,     queueDecoder,+    unvalidatedQueueDecoder,     queueMetricsDecoder,     -- Topic decoders (pgmq 1.11.0+)     topicBindingDecoder,@@ -11,7 +12,9 @@   ) where +import Data.Aeson qualified as Aeson import Data.Bifunctor (first)+import Data.Maybe (fromMaybe) import Data.Text (pack) import Hasql.Decoders qualified as D import Pgmq.Hasql.Statements.Types (QueueMetrics (..))@@ -24,12 +27,21 @@     RoutingMatch (..),     TopicBinding (..),     TopicSendResult (..),+    UnvalidatedQueue (..),     parseQueueName,     parseTopicPattern,   )  -- | Decoder for pgmq.message_record type -- Column order matches pgmq SQL: msg_id, read_ct, enqueued_at, last_read_at, vt, message, headers+--+-- The @message@ column is nullable in the queue table, and+-- @pgmq.send(queue, NULL::jsonb)@ is legal SQL any non-Haskell producer can+-- issue. A SQL NULL body decodes as JSON @null@ (@MessageBody Aeson.Null@) —+-- an accepted conflation with an explicitly-sent JSON @null@ body, since both+-- mean \"no usable payload\". Requiring a non-null cell here would instead+-- fail the whole batch at decode, after the read statement had already bumped+-- @vt@ and @read_ct@ for every message in it, leaving an invisible poison row. messageDecoder :: D.Row Message messageDecoder =   ( \msgId readCt enqueuedAt lastReadAt vt body headers ->@@ -48,7 +60,7 @@     <*> D.column (D.nonNullable D.timestamptz) -- enqueued_at     <*> D.column (D.nullable D.timestamptz) -- last_read_at     <*> D.column (D.nonNullable D.timestamptz) -- vt-    <*> (MessageBody <$> D.column (D.nonNullable D.jsonb)) -- message+    <*> (MessageBody . fromMaybe Aeson.Null <$> D.column (D.nullable D.jsonb)) -- message (SQL NULL -> JSON null)     <*> D.column (D.nullable D.jsonb) -- headers  messageIdDecoder :: D.Row MessageId@@ -60,6 +72,25 @@ queueDecoder =   (\name isPartitioned isUnlogged createdAt -> Queue name createdAt isPartitioned isUnlogged)     <$> D.column (D.nonNullable $ D.refine (first (pack . show) . parseQueueName) D.varchar)+    <*> D.column (D.nonNullable D.bool)+    <*> D.column (D.nonNullable D.bool)+    <*> D.column (D.nonNullable D.timestamptz)++-- | Like 'queueDecoder' but with the queue name left as plain text.+--+-- The server's only queue-name check is length, so any client sharing the+-- database can create a name 'parseQueueName' rejects. 'queueDecoder' refines+-- that column and therefore fails the entire listing on one such row; this+-- decoder does not, so state inspection can observe foreign queues.+--+-- Column order matches 'queueDecoder': queue_name (varchar), is_partitioned+-- (bool), is_unlogged (bool), created_at (timestamptz).+unvalidatedQueueDecoder :: D.Row UnvalidatedQueue+unvalidatedQueueDecoder =+  ( \name isPartitioned isUnlogged createdAt ->+      UnvalidatedQueue name createdAt isPartitioned isUnlogged+  )+    <$> D.column (D.nonNullable D.varchar)     <*> D.column (D.nonNullable D.bool)     <*> D.column (D.nonNullable D.bool)     <*> D.column (D.nonNullable D.timestamptz)

@@ -140,12 +140,13 @@     <> (view #messageHeaders >$< E.param (E.nonNullable (E.array (E.dimension foldl' (E.element (E.nonNullable messageHeadersValue))))))     <> (view #scheduledAt >$< E.param (E.nonNullable E.timestamptz)) --- | Encoder for the 3-param pgmq.read (without conditional filter)+-- | Encoder for the 4-param pgmq.read (including the conditional filter) readMessageEncoder :: E.Params ReadMessage readMessageEncoder =   (view #queueName >$< E.param (E.nonNullable queueNameValue))     <> (view #delay >$< E.param (E.nonNullable E.int4))     <> (view #batchSize >$< E.param (E.nullable E.int4))+    <> (view #conditional >$< E.param (E.nullable E.jsonb))  -- | Encoder for PopMessage (pgmq 1.7.0+) popMessageEncoder :: E.Params PopMessage

@@ -25,6 +25,8 @@     setVisibilityTimeoutAt,     batchSetVisibilityTimeoutAt,     listQueues,+    listQueuesUnvalidated,+    listFifoIndexQueueNames,     pop,     queueMetrics,     allQueueMetrics,@@ -109,6 +111,7 @@     TopicBinding,     TopicPattern,     TopicSendResult,+    UnvalidatedQueue,   )  createQueue :: QueueName -> Session ()@@ -160,15 +163,18 @@ deleteAllMessagesFromQueue :: QueueName -> Session Int64 deleteAllMessagesFromQueue qname = statement qname Msg.deleteAllMessagesFromQueue -changeVisibilityTimeout :: VisibilityTimeoutQuery -> Session Message+-- | Returns Nothing when the message no longer exists (already deleted, archived,+-- or popped) rather than failing the session.+changeVisibilityTimeout :: VisibilityTimeoutQuery -> Session (Maybe Message) changeVisibilityTimeout query = statement query Msg.changeVisibilityTimeout  -- | Batch update visibility timeout (pgmq 1.8.0+) batchChangeVisibilityTimeout :: BatchVisibilityTimeoutQuery -> Session (Vector Message) batchChangeVisibilityTimeout query = statement query Msg.batchChangeVisibilityTimeout --- | Set visibility timeout to an absolute timestamp (pgmq 1.10.0+)-setVisibilityTimeoutAt :: VisibilityTimeoutAtQuery -> Session Message+-- | Set visibility timeout to an absolute timestamp (pgmq 1.10.0+).+-- Returns Nothing when the message no longer exists.+setVisibilityTimeoutAt :: VisibilityTimeoutAtQuery -> Session (Maybe Message) setVisibilityTimeoutAt query = statement query Msg.setVisibilityTimeoutAt  -- | Batch set visibility timeout to an absolute timestamp (pgmq 1.10.0+)@@ -177,6 +183,17 @@  listQueues :: Session [Queue] listQueues = statement () Stmt.listQueues++-- | Like 'listQueues' but with queue names left unvalidated, so a queue+-- created by another client under a name 'Pgmq.Types.parseQueueName' rejects+-- does not fail the whole listing.+listQueuesUnvalidated :: Session [UnvalidatedQueue]+listQueuesUnvalidated = statement () Stmt.listQueuesUnvalidated++-- | Queue names that already have their FIFO headers index, read from the+-- @pg_indexes@ catalog view (pgmq has no index-existence function).+listFifoIndexQueueNames :: Session [Text]+listFifoIndexQueueNames = statement () Stmt.listFifoIndexQueueNames  createPartitionedQueue :: CreatePartitionedQueue -> Session () createPartitionedQueue q = statement q Stmt.createPartitionedQueue

@@ -162,14 +162,19 @@     decoder = D.rowList messageIdDecoder  -- | https://pgmq.github.io/pgmq/api/sql/functions/#read--- Note: conditional parameter added in pgmq 1.5.0--- We use the 3-param version since the 4-param version fails with NULL conditional--- (message @> NULL = NULL, not TRUE, so no rows match).--- To use conditional filtering, use readMessageConditional instead.+-- Note: conditional parameter added in pgmq 1.5.0. It is a JSONB containment+-- filter: a message is returned only when @message \@> conditional@ holds.+-- Nothing (equivalently '{}'::jsonb) means no filtering, so the coalesce+-- neutralizes an unbound filter without changing which rows match.+--+-- The coalesce on the batch size is load-bearing: a bound SQL NULL never+-- triggers a plpgsql parameter DEFAULT (defaults apply only to omitted+-- arguments), and NULL reaching the LIMIT clause inside pgmq.read means+-- LIMIT ALL — which would lease the entire queue in one call. readMessage :: Statement ReadMessage (Vector Message) readMessage = preparable sql readMessageEncoder decoder   where-    sql = "select * from pgmq.read($1,$2,$3)"+    sql = "select * from pgmq.read($1,$2,coalesce($3,1),coalesce($4,'{}'::jsonb))"     decoder = D.rowVector messageDecoder  -- | https://pgmq.github.io/pgmq/api/sql/functions/#delete-single@@ -208,13 +213,18 @@     sql = "select * from pgmq.purge_queue($1)"     decoder = D.singleRow $ D.column $ D.nonNullable D.int8 --- | Sets the visibility timeout of a message to a specified time duration in the future. Returns the record of the message that was updated.+-- | Sets the visibility timeout of a message to a specified time duration in the future.+-- Returns the record of the message that was updated, or Nothing when the message no+-- longer exists (already deleted, archived, or popped).+--+-- pgmq.set_vt is RETURNS SETOF and yields zero rows for an absent msg_id, which is an+-- ordinary outcome when another consumer raced ahead — not an infrastructure failure. -- | https://pgmq.github.io/pgmq/api/sql/functions/#set_vt-changeVisibilityTimeout :: Statement VisibilityTimeoutQuery Message+changeVisibilityTimeout :: Statement VisibilityTimeoutQuery (Maybe Message) changeVisibilityTimeout = preparable sql visibilityTimeoutQueryEncoder decoder   where     sql = "select * from pgmq.set_vt($1,$2,$3)"-    decoder = D.singleRow messageDecoder+    decoder = D.rowMaybe messageDecoder  -- | Batch update visibility timeout for multiple messages (pgmq 1.8.0+) -- | https://pgmq.github.io/pgmq/api/sql/functions/#set_vt@@ -225,12 +235,14 @@     decoder = D.rowVector messageDecoder  -- | Set visibility timeout to an absolute timestamp (pgmq 1.10.0+)+-- Returns Nothing when the message no longer exists (already deleted, archived, or+-- popped) — see 'changeVisibilityTimeout' for why that is not an error. -- | https://pgmq.github.io/pgmq/api/sql/functions/#set_vt-setVisibilityTimeoutAt :: Statement VisibilityTimeoutAtQuery Message+setVisibilityTimeoutAt :: Statement VisibilityTimeoutAtQuery (Maybe Message) setVisibilityTimeoutAt = preparable sql visibilityTimeoutAtQueryEncoder decoder   where     sql = "select * from pgmq.set_vt($1,$2,$3)"-    decoder = D.singleRow messageDecoder+    decoder = D.rowMaybe messageDecoder  -- | Batch set visibility timeout to an absolute timestamp (pgmq 1.10.0+) -- | https://pgmq.github.io/pgmq/api/sql/functions/#set_vt@@ -241,19 +253,28 @@     decoder = D.rowVector messageDecoder  -- | https://pgmq.github.io/pgmq/api/sql/functions/#read_with_poll+-- Shares readMessage's coalesce rationale: a NULL batch size would become+-- LIMIT ALL inside the polling loop and lease the whole queue, and a NULL+-- conditional is normalized to the no-filter '{}' value. readWithPoll :: Statement ReadWithPollMessage (Vector Message) readWithPoll = preparable sql readWithPollEncoder decoder   where-    sql = "select * from pgmq.read_with_poll($1,$2,$3,$4,$5,$6)"+    sql = "select * from pgmq.read_with_poll($1,$2,coalesce($3,1),$4,$5,coalesce($6,'{}'::jsonb))"     decoder = D.rowVector messageDecoder  -- | Pop messages from queue (atomic read + delete) -- https://pgmq.github.io/pgmq/api/sql/functions/#pop -- Note: qty parameter added in pgmq 1.7.0+--+-- The coalesce is what makes "Nothing = 1" true. A bound SQL NULL never+-- triggers the plpgsql DEFAULT of 1 (defaults apply only to omitted+-- arguments), and NULL in a LIMIT clause means LIMIT ALL — so without it,+-- popping with no explicit quantity would delete and return the entire+-- queue in a single statement, with no visibility timeout to fall back on. pop :: Statement PopMessage (Vector Message) pop = preparable sql popMessageEncoder decoder   where-    sql = "select * from pgmq.pop($1,$2)"+    sql = "select * from pgmq.pop($1,coalesce($2,1))"     decoder = D.rowVector messageDecoder  -- | FIFO read - fills batch from same message group (pgmq 1.8.0+)

@@ -56,11 +56,22 @@     sql = "select from pgmq.detach_archive($1)"  -- | Enable insert notifications for a queue (pgmq 1.7.0+)--- Notifications are sent via PostgreSQL LISTEN/NOTIFY to channel pgmq_<queue_name>+--+-- Notifications are sent via PostgreSQL LISTEN\/NOTIFY on the channel computed by+-- 'Pgmq.Types.notifyChannelName' — @pgmq.q_\<lowercased queue name\>.INSERT@. The+-- name contains dots, so LISTEN requires it double-quoted. Use the helper rather+-- than assembling the name by hand.+--+-- NOTIFY is fire-and-forget: notifications are not queued for disconnected+-- listeners, and a configured throttle interval suppresses them by design. Every+-- consumer needs a poll fallback in addition to LISTEN. enableNotifyInsert :: Statement EnableNotifyInsert () enableNotifyInsert = preparable sql enableNotifyInsertEncoder D.noResult   where-    sql = "select from pgmq.enable_notify_insert($1, $2)"+    -- The coalesce makes "Nothing = 250ms" true. throttle_interval_ms is NOT NULL+    -- with a column DEFAULT, but a column DEFAULT does not apply to an explicitly+    -- supplied NULL, so a bound SQL NULL raised SQLSTATE 23502 on every call.+    sql = "select from pgmq.enable_notify_insert($1, coalesce($2, 250))"  -- | Disable insert notifications for a queue disableNotifyInsert :: Statement QueueName ()

@@ -1,17 +1,20 @@ module Pgmq.Hasql.Statements.QueueObservability   ( listQueues,+    listQueuesUnvalidated,+    listFifoIndexQueueNames,     queueMetrics,     allQueueMetrics,   ) where +import Data.Text (Text) import Hasql.Decoders qualified as D import Hasql.Encoders qualified as E import Hasql.Statement (Statement, preparable)-import Pgmq.Hasql.Decoders (queueDecoder, queueMetricsDecoder)+import Pgmq.Hasql.Decoders (queueDecoder, queueMetricsDecoder, unvalidatedQueueDecoder) import Pgmq.Hasql.Encoders (queueNameEncoder) import Pgmq.Hasql.Statements.Types (QueueMetrics)-import Pgmq.Types (Queue, QueueName)+import Pgmq.Types (Queue, QueueName, UnvalidatedQueue)  -- | List all queues that currently exist -- | https://pgmq.github.io/pgmq/api/sql/functions/#list_queues@@ -20,6 +23,32 @@   where     sql = "select * from pgmq.list_queues()"     decoder = D.rowList queueDecoder++-- | Like 'listQueues' but with names left unvalidated, so rows created by+-- other clients with names 'Pgmq.Types.parseQueueName' rejects still decode.+-- | https://pgmq.github.io/pgmq/api/sql/functions/#list_queues+listQueuesUnvalidated :: Statement () [UnvalidatedQueue]+listQueuesUnvalidated = preparable sql E.noParams decoder+  where+    sql = "select * from pgmq.list_queues()"+    decoder = D.rowList unvalidatedQueueDecoder++-- | Queue names (in the lowercased physical form pgmq derives table names from)+-- that already carry the FIFO headers index @q_\<name\>_fifo_idx@.+--+-- Unlike every other statement in this module this reads a PostgreSQL catalog+-- view rather than calling a @pgmq.*@ function, because pgmq exposes no+-- index-existence query: @pgmq.create_fifo_index@ delegates to+-- @CREATE INDEX IF NOT EXISTS@ and reports nothing back. A caller that wants to+-- say truthfully whether it created an index has to look in @pg_indexes@.+listFifoIndexQueueNames :: Statement () [Text]+listFifoIndexQueueNames = preparable sql E.noParams decoder+  where+    sql =+      "select substring(indexname from '^q_(.*)_fifo_idx$')::text \+      \from pg_indexes \+      \where schemaname = 'pgmq' and indexname ~ '^q_.*_fifo_idx$'"+    decoder = D.rowList (D.column (D.nonNullable D.text))  -- | https://pgmq.github.io/pgmq/api/sql/functions/#metrics queueMetrics :: Statement QueueName QueueMetrics

@@ -154,8 +154,13 @@ data ReadMessage = ReadMessage   { queueName :: !QueueName,     delay :: !Delay,+    -- | Number of messages to read. Nothing = 1, applied via COALESCE in the+    -- statement (a bound SQL NULL never triggers the plpgsql DEFAULT, and NULL+    -- in a LIMIT clause means LIMIT ALL).     batchSize :: !(Maybe Int32),-    -- | Optional JSONB filter (pgmq 1.5.0+)+    -- | Optional JSONB containment filter (pgmq 1.5.0+). A message is returned+    -- only when its body contains this object (SQL @message \@> conditional@).+    -- Nothing means no filtering.     conditional :: !(Maybe Value)   }   deriving stock (Generic)@@ -163,9 +168,13 @@ data ReadWithPollMessage = ReadWithPollMessage   { queueName :: !QueueName,     delay :: !Delay,+    -- | Number of messages to read. Nothing = 1, applied via COALESCE in the+    -- statement, for the same reason as 'ReadMessage'.     batchSize :: !(Maybe Int32),     maxPollSeconds :: !Int32,     pollIntervalMs :: !Int32,+    -- | Optional JSONB containment filter (pgmq 1.5.0+). Nothing means no+    -- filtering.     conditional :: !(Maybe Value)   }   deriving stock (Generic)@@ -173,7 +182,10 @@ -- | Parameters for popping messages from a queue (pgmq 1.7.0+) data PopMessage = PopMessage   { queueName :: !QueueName,-    -- | Number of messages to pop (Nothing = default 1)+    -- | Number of messages to pop. Nothing = 1, applied via COALESCE in the+    -- statement (a bound SQL NULL never triggers the plpgsql DEFAULT, and NULL+    -- in a LIMIT clause means LIMIT ALL — which for pop would delete the whole+    -- queue).     qty :: !(Maybe Int32)   }   deriving stock (Generic)@@ -181,7 +193,9 @@ -- | Enable queue notifications (pgmq 1.7.0+, throttling in 1.8.0+) data EnableNotifyInsert = EnableNotifyInsert   { queueName :: !QueueName,-    -- | Minimum ms between notifications (Nothing = default 250ms)+    -- | Minimum ms between notifications. Nothing = 250ms, applied via COALESCE+    -- in the statement so a bound SQL NULL never reaches the NOT NULL column+    -- (a column DEFAULT does not apply to an explicitly supplied NULL).     throttleIntervalMs :: !(Maybe Int32)   }   deriving stock (Generic)

@@ -33,7 +33,7 @@ import Pgmq.Types qualified as PgmqTypes import Test.Tasty (TestTree, testGroup) import Test.Tasty.HUnit (assertBool, assertEqual, testCase)-import TestUtils (assertSession, cleanupQueue)+import TestUtils (assertJust, assertSession, cleanupQueue)  -- | All advanced operation tests tests :: Pool.Pool -> TestTree@@ -180,7 +180,8 @@               messageId = msgId,               visibilityTime = futureTime             }-    updated <- assertSession pool (Sessions.setVisibilityTimeoutAt vtQuery)+    -- The message exists, so set_vt must return Just it.+    updated <- assertJust =<< assertSession pool (Sessions.setVisibilityTimeoutAt vtQuery)     assertEqual "Should return the updated message" msgId (PgmqTypes.messageId updated)     cleanupQueue pool queueName 

@@ -0,0 +1,228 @@+{-# LANGUAGE OverloadedStrings #-}++-- | PGH-7 evidence: what mixed-case queue names do to the SQL layer.+--+-- pgmq's SQL is consistent-by-lowercasing for /physical/ names —+-- @pgmq.format_table_name@ lowercases — but @pgmq.meta@ stores the caller's+-- /original/ casing, and the notify trigger extracts the /lowercased/ name from+-- the physical table it fires on. Three views of one name that only agree for+-- lowercase input. These tests drive the SQL layer directly (raw statements,+-- never the Haskell API) and document the consequences:+--+--   1. @create('MyQueue')@ then @create('myqueue')@ yields ONE physical table+--      with TWO meta rows: two logical queues silently interleaving in one+--      table.+--   2. @drop_queue('myqueue')@ destroys the mixed-case alias's messages while+--      its meta row lives on, pointing at nothing.+--   3. @enable_notify_insert('MyQueue')@ writes a throttle row the trigger's+--      lowercase lookup never matches, so the configured throttle interval is+--      silently ignored. (Since migration 0003 the trigger fails open on a+--      missing row, so notifications fire /unthrottled/; before it, they never+--      fired at all. Either way the configuration is dead on arrival.)+--+-- These are evidence tests: they pass against the current SQL layer and stay+-- green after the Haskell boundary starts rejecting mixed-case names, at which+-- point the states they construct become unreachable from validated input.+-- They run on a dedicated PostgreSQL instance, never the suite-shared one,+-- because a mixed-case @pgmq.meta@ row poisons @listQueues@ decoding for every+-- concurrent test once @parseQueueName@ rejects it.+module AliasingSpec (tests) where++import Control.Monad (void)+import Data.Int (Int64)+import Data.List.NonEmpty (NonEmpty (..))+import Data.Text (Text)+import Data.Text qualified as T+import Data.Vector qualified as V+import Data.Word (Word32)+import Database.PostgreSQL.Migrate+  ( defaultRunOptions,+    migrationPlan,+    runMigrationPlan,+  )+import EphemeralPg qualified as Pg+import Hasql.Decoders qualified as D+import Hasql.Pool qualified as Pool+import Hasql.Pool.Config qualified as PoolConfig+import Hasql.Session (Session, statement)+import Hasql.Statement (unpreparable)+import Pgmq.Migration qualified as Migration+import System.Random (randomRIO)+import Test.Tasty (TestTree, testGroup, withResource)+import Test.Tasty.HUnit (assertBool, assertEqual, assertFailure, testCase)++tests :: TestTree+tests =+  withResource acquireDb releaseDb $ \getDb ->+    testGroup+      "Mixed-Case Queue Aliasing (PGH-7 evidence)"+      [ testOneTableTwoMetaRows getDb,+        testDropDestroysTheAlias getDb,+        testNotifyThrottleNeverMatches getDb+      ]++-- | Both casings create the same physical table, interleave their messages in+-- it, and leave two rows in @pgmq.meta@.+testOneTableTwoMetaRows :: IO (Pg.Database, Pool.Pool) -> TestTree+testOneTableTwoMetaRows getDb = testCase "create in both casings yields one physical table and two meta rows" $ do+  (_, pool) <- getDb+  (mixed, lower) <- genQueuePair+  assertSession pool (rawUnit ("select pgmq.create('" <> mixed <> "')"))+  assertSession pool (rawUnit ("select pgmq.create('" <> lower <> "')"))+  -- Exactly one physical table exists for the pair, and it is the lowercase+  -- one. The case-insensitive count would catch a hypothetical q_MyQueue_<n>.+  tables <-+    assertSession pool $+      rawCount ("select count(*) from pg_tables where schemaname = 'pgmq' and lower(tablename) = 'q_" <> lower <> "'")+  assertEqual "Exactly one physical table for both casings" 1 tables+  lowerTables <-+    assertSession pool $+      rawCount ("select count(*) from pg_tables where schemaname = 'pgmq' and tablename = 'q_" <> lower <> "'")+  assertEqual "The one physical table is the lowercased name" 1 lowerTables+  metas <-+    assertSession pool $+      rawCount ("select count(*) from pgmq.meta where lower(queue_name) = '" <> lower <> "'")+  assertEqual "Two meta rows share the one physical table" 2 metas+  -- A message sent through the mixed-case name is read back through the+  -- lowercase name: the \"two\" queues interleave in one table.+  void $ assertSession pool (rawIds ("select pgmq.send('" <> mixed <> "', '{\"via\":\"upper\"}'::jsonb)"))+  void $ assertSession pool (rawIds ("select pgmq.send('" <> lower <> "', '{\"via\":\"lower\"}'::jsonb)"))+  readBack <-+    assertSession pool $+      rawCount ("select count(*) from pgmq.read('" <> lower <> "', 0, 10)")+  assertEqual "Reading via the lowercase name returns both casings' messages" 2 readBack+  -- Cleanup: drop the lowercase queue (table and meta row), then remove the+  -- orphaned mixed-case meta row directly — drop_queue refuses once the table+  -- is gone.+  void $ assertSession pool (rawBool ("select pgmq.drop_queue('" <> lower <> "')"))+  assertSession pool (rawUnit ("delete from pgmq.meta where queue_name = '" <> mixed <> "'"))++-- | Dropping the lowercase twin destroys the mixed-case alias's messages; the+-- alias's meta row survives, pointing at a table that no longer exists.+testDropDestroysTheAlias :: IO (Pg.Database, Pool.Pool) -> TestTree+testDropDestroysTheAlias getDb = testCase "drop_queue on one casing breaks the other" $ do+  (_, pool) <- getDb+  (mixed, lower) <- genQueuePair+  assertSession pool (rawUnit ("select pgmq.create('" <> mixed <> "')"))+  assertSession pool (rawUnit ("select pgmq.create('" <> lower <> "')"))+  void $ assertSession pool (rawIds ("select pgmq.send('" <> mixed <> "', '{\"owner\":\"mixed\"}'::jsonb)"))+  dropped <- assertSession pool (rawBool ("select pgmq.drop_queue('" <> lower <> "')"))+  assertBool "drop_queue on the lowercase twin reports success" dropped+  -- The mixed-case alias is now broken: its meta row survives but every send+  -- through it fails on the missing physical table (SQLSTATE 42P01).+  sendResult <- Pool.use pool (rawIds ("select pgmq.send('" <> mixed <> "', '{\"after\":\"drop\"}'::jsonb)"))+  case sendResult of+    Right _ -> assertFailure "Sending via the mixed-case alias should fail once the twin is dropped"+    Left err ->+      assertBool+        ("Expected undefined_table (42P01), got: " <> show err)+        ("42P01" `T.isInfixOf` T.pack (show err))+  survivors <-+    assertSession pool $+      rawCount ("select count(*) from pgmq.meta where queue_name = '" <> mixed <> "'")+  assertEqual "The mixed-case meta row survives the drop" 1 survivors+  assertSession pool (rawUnit ("delete from pgmq.meta where queue_name = '" <> mixed <> "'"))++-- | @enable_notify_insert@ on a mixed-case name writes a throttle row keyed by+-- the original casing, but the trigger fires on the physical table and looks up+-- the /lowercased/ name — so the row is never matched and the configured+-- throttle never applies. Since migration 0003 the trigger fails open on the+-- missing row (notifying unthrottled); before it, the same mismatch silently+-- suppressed every notification.+testNotifyThrottleNeverMatches :: IO (Pg.Database, Pool.Pool) -> TestTree+testNotifyThrottleNeverMatches getDb = testCase "mixed-case enable_notify_insert configures a throttle the trigger never matches" $ do+  (_, pool) <- getDb+  (mixed, _) <- genQueuePair+  control <- genControlName+  assertSession pool (rawUnit ("select pgmq.create('" <> mixed <> "')"))+  assertSession pool (rawUnit ("select pgmq.enable_notify_insert('" <> mixed <> "', 60000)"))+  assertSession pool (rawUnit ("select pgmq.create('" <> control <> "')"))+  assertSession pool (rawUnit ("select pgmq.enable_notify_insert('" <> control <> "', 60000)"))+  void $ assertSession pool (rawIds ("select pgmq.send('" <> mixed <> "', '{\"probe\":\"mixed\"}'::jsonb)"))+  void $ assertSession pool (rawIds ("select pgmq.send('" <> control <> "', '{\"probe\":\"control\"}'::jsonb)"))+  mixedFrozen <-+    assertSession pool $+      rawBool+        ( "select last_notified_at = to_timestamp(0) from pgmq.notify_insert_throttle where queue_name = '"+            <> mixed+            <> "'"+        )+  assertBool+    "The mixed-case throttle row is never matched: last_notified_at stays at the epoch"+    mixedFrozen+  controlFrozen <-+    assertSession pool $+      rawBool+        ( "select last_notified_at = to_timestamp(0) from pgmq.notify_insert_throttle where queue_name = '"+            <> control+            <> "'"+        )+  assertBool+    "The lowercase control's throttle row is matched and stamped"+    (not controlFrozen)+  void $ assertSession pool (rawBool ("select pgmq.drop_queue('" <> control <> "')"))+  void $ assertSession pool (rawBool ("select pgmq.drop_queue('" <> mixed <> "')"))++-- Dedicated database plumbing -------------------------------------------------++-- | Start a dedicated PostgreSQL instance with the full pgmq migration ledger+-- installed, exactly as @EphemeralDb@ does for the shared one.+acquireDb :: IO (Pg.Database, Pool.Pool)+acquireDb = do+  started <- Pg.startCached Pg.defaultConfig Pg.defaultCacheConfig+  db <- either (\err -> error ("could not start a dedicated PostgreSQL: " <> show err)) pure started+  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 ()+  pool <-+    Pool.acquire $+      PoolConfig.settings+        [ PoolConfig.size 2,+          PoolConfig.staticConnectionSettings (Pg.connectionSettings db)+        ]+  pure (db, pool)++releaseDb :: (Pg.Database, Pool.Pool) -> IO ()+releaseDb (db, pool) = do+  Pool.release pool+  Pg.stop db++-- Raw statement helpers -------------------------------------------------------+--+-- Queue names are spliced into the SQL text because these tests must construct+-- names the Haskell API (rightly) refuses. Every spliced value is generated+-- below from @[A-Za-z0-9_]@, so splicing is safe here.++rawUnit :: Text -> Session ()+rawUnit sqlText = statement () (unpreparable sqlText mempty D.noResult)++rawCount :: Text -> Session Int64+rawCount sqlText = statement () (unpreparable sqlText mempty (D.singleRow (D.column (D.nonNullable D.int8))))++rawBool :: Text -> Session Bool+rawBool sqlText = statement () (unpreparable sqlText mempty (D.singleRow (D.column (D.nonNullable D.bool))))++rawIds :: Text -> Session (V.Vector Int64)+rawIds sqlText = statement () (unpreparable sqlText mempty (D.rowVector (D.column (D.nonNullable D.int8))))++assertSession :: Pool.Pool -> Session a -> IO a+assertSession pool session = do+  result <- Pool.use pool session+  case result of+    Left err -> assertFailure $ "Session failed: " <> show err+    Right a -> pure a++-- | A mixed-case name and its lowercase twin, sharing one random suffix.+genQueuePair :: IO (Text, Text)+genQueuePair = do+  suffix <- randomRIO (10000 :: Word32, 99999)+  let lower = "myqueue_" <> T.pack (show suffix)+  pure ("MyQueue_" <> T.pack (show suffix), lower)++genControlName :: IO Text+genControlName = do+  suffix <- randomRIO (10000 :: Word32, 99999)+  pure ("ctrl_" <> T.pack (show suffix))

@@ -22,7 +22,7 @@ import Pgmq.Types qualified import Test.Tasty (TestTree, testGroup) import Test.Tasty.HUnit (assertBool, assertEqual, testCase)-import TestUtils (assertSession, cleanupQueue)+import TestUtils (assertJust, assertSession, cleanupQueue)  -- | All per-function decoder tests tests :: Pool.Pool -> TestTree@@ -112,7 +112,9 @@               messageId = msgId,               visibilityTimeoutOffset = 60             }-    msg <- assertSession pool (Sessions.changeVisibilityTimeout vtQuery)+    -- The message exists, so set_vt must return Just it (Nothing means the row+    -- was raced away, which cannot happen here).+    msg <- assertJust =<< assertSession pool (Sessions.changeVisibilityTimeout vtQuery)      -- Verify message fields     assertBool "messageId should be positive" (unMessageId (Pgmq.Types.messageId msg) > 0)

@@ -11,6 +11,7 @@     withTestFixture,      -- * Re-exports+    Database,     StartError,   ) where@@ -24,7 +25,8 @@     runMigrationPlan,   ) import EphemeralPg-  ( StartError,+  ( Database,+    StartError,     connectionSettings,     withCached,   )@@ -34,8 +36,10 @@ import Pgmq.Types (QueueName, parseQueueName) import System.Random (randomRIO) --- | Run an action with a temporary PostgreSQL database that has pgmq schema installed-withPgmqDb :: (Pool.Pool -> IO a) -> IO (Either StartError a)+-- | Run an action with a temporary PostgreSQL database that has pgmq schema installed.+-- The 'Database' handle is passed alongside the pool because tests that need a raw+-- libpq connection (LISTEN\/NOTIFY has no hasql API) need its connection string.+withPgmqDb :: (Pool.Pool -> Database -> IO a) -> IO (Either StartError a) withPgmqDb action = withCached $ \db -> do   let connSettings = connectionSettings db       poolConfig =@@ -49,12 +53,12 @@   installResult <- runMigrationPlan defaultRunOptions connSettings plan   case installResult of     Left migrationErr -> error $ "Migration failed: " <> show migrationErr-    Right _ -> action pool+    Right _ -> action pool db  -- | Run an action with a connection pool to a temporary PostgreSQL database -- The database will have the pgmq schema installed withPgmqPool :: (Pool.Pool -> IO a) -> IO (Either StartError a)-withPgmqPool = withPgmqDb+withPgmqPool action = withPgmqDb (\pool _ -> action pool)  -- | Test fixture with isolated queue for a test data TestFixture = TestFixture

@@ -3,10 +3,16 @@ module Main (main) where  import AdvancedOpsSpec qualified+import AliasingSpec qualified import AllFunctionsDecoderSpec qualified import DecoderValidationSpec qualified-import EphemeralDb (withPgmqPool)+import EphemeralDb (withPgmqDb) import MessageSpec qualified+import MixedCaseRemediationSpec qualified+import NotifyChannelSpec qualified+import NotifyRaceSpec qualified+import NullBodySpec qualified+import NullSemanticsSpec qualified import QueueSpec qualified import RoundTripSpec qualified import SchemaSpec qualified@@ -16,13 +22,26 @@ main :: IO () main = do   -- Run tests with a shared temporary database-  result <- withPgmqPool $ \pool -> do+  result <- withPgmqDb $ \pool db -> do     let tree =           testGroup             "pgmq-hasql"             [ QueueSpec.tests pool,               MessageSpec.tests pool,               AdvancedOpsSpec.tests pool,+              NullSemanticsSpec.tests pool,+              NullBodySpec.tests pool,+              NotifyRaceSpec.tests pool,+              -- Both construct mixed-case pgmq.meta rows, which poison+              -- listQueues decoding for every concurrent test — so each runs+              -- on its own dedicated PostgreSQL instance, never the shared+              -- pool. They are separate instances because the remediation+              -- sweeps every mixed-case row in its database.+              AliasingSpec.tests,+              MixedCaseRemediationSpec.tests,+              -- Needs the Database handle: LISTEN/NOTIFY has no hasql API, so+              -- the round-trip test opens a raw libpq connection.+              NotifyChannelSpec.tests pool db,               SchemaSpec.tests pool,               RoundTripSpec.tests pool,               DecoderValidationSpec.tests pool,

@@ -21,7 +21,7 @@ import Pgmq.Types qualified as PgmqTypes import Test.Tasty (TestTree, testGroup) import Test.Tasty.HUnit (assertBool, assertEqual, testCase)-import TestUtils (assertSession, cleanupQueue)+import TestUtils (assertJust, assertSession, cleanupQueue)  -- | All message operation tests tests :: Pool.Pool -> TestTree@@ -243,7 +243,8 @@               messageId = msgId,               visibilityTimeoutOffset = 60             }-    msg <- assertSession pool (Sessions.changeVisibilityTimeout vtQuery)+    -- The message exists, so set_vt must return Just it.+    msg <- assertJust =<< assertSession pool (Sessions.changeVisibilityTimeout vtQuery)     assertEqual "Should return the message" msgId (PgmqTypes.messageId msg)     cleanupQueue pool queueName 

@@ -0,0 +1,358 @@+{-# LANGUAGE OverloadedStrings #-}++-- | PGH-7: the documented mixed-case remediation must preserve topic bindings+-- and notification configuration, transactionally, must be safe to rerun, and+-- must delete an orphaned row (physical table already destroyed) rather than+-- resurrect it as a phantom queue.+--+-- The stricter 'Pgmq.Types.parseQueueName' makes pre-existing mixed-case+-- @pgmq.meta@ rows fail @listQueues@ decoding, so deployments must remediate+-- before upgrading. The remediation (canonical copy in+-- @docs\/design\/016-queue-name-validation.md@) cannot be a naive rename or+-- delete: both foreign keys onto @pgmq.meta (queue_name)@ — from+-- @pgmq.topic_bindings@ and @pgmq.notify_insert_throttle@ — lack @ON UPDATE@+-- and carry @ON DELETE CASCADE@, so an UPDATE of a referenced parent fails and+-- a DELETE silently destroys routing and notification configuration.+--+-- This module runs on its own dedicated PostgreSQL instance: the remediation+-- sweeps every mixed-case row in the database, so it must never share an+-- instance with other tests that construct mixed-case rows (AliasingSpec), let+-- alone the suite-shared pool. For the same reason its own cases run+-- sequentially — each one executes the global sweep, which would otherwise+-- race a sibling's setup.+module MixedCaseRemediationSpec (tests) where++import Control.Monad (void)+import Data.Int (Int64)+import Data.List.NonEmpty (NonEmpty (..))+import Data.Text (Text)+import Data.Text qualified as T+import Data.Vector qualified as V+import Data.Word (Word32)+import Database.PostgreSQL.Migrate+  ( defaultRunOptions,+    migrationPlan,+    runMigrationPlan,+  )+import EphemeralPg qualified as Pg+import Hasql.Decoders qualified as D+import Hasql.Pool qualified as Pool+import Hasql.Pool.Config qualified as PoolConfig+import Hasql.Session (Session, statement)+import Hasql.Statement (unpreparable)+import Pgmq.Migration qualified as Migration+import System.Random (randomRIO)+import Test.Tasty (DependencyType (AllFinish), TestTree, sequentialTestGroup, withResource)+import Test.Tasty.HUnit (assertBool, assertEqual, assertFailure, testCase)++-- Sequential, not parallel: every test runs the remediation, and the+-- remediation sweeps EVERY mixed-case row in the database — a concurrent+-- sibling's sweep landing between this test's @create@ and its @bind_topic@+-- deletes the parent row out from under the binding (23503).+tests :: TestTree+tests =+  withResource acquireDb releaseDb $ \getDb ->+    sequentialTestGroup+      "Mixed-Case Remediation (PGH-7)"+      AllFinish+      [ testNoTwinRename getDb,+        testTwinMerge getDb,+        testOrphanDeletion getDb+      ]++-- | The documented remediation, verbatim from design note 016. One DO block =+-- one transaction; rerunning it after success is a no-op because the driving+-- query returns no rows.+remediationSql :: Text+remediationSql =+  T.unlines+    [ "DO $remediate$",+      "DECLARE",+      "  bad RECORD;",+      "  twin_exists BOOLEAN;",+      "  table_exists BOOLEAN;",+      "BEGIN",+      "  FOR bad IN",+      "    SELECT m.queue_name AS mixed_name, lower(m.queue_name) AS canonical_name",+      "    FROM pgmq.meta m",+      "    WHERE m.queue_name <> lower(m.queue_name)",+      "  LOOP",+      "    PERFORM pgmq.acquire_queue_lock(bad.mixed_name);",+      "    PERFORM pgmq.acquire_queue_lock(bad.canonical_name);",+      "    PERFORM 1 FROM pgmq.meta",+      "      WHERE queue_name IN (bad.mixed_name, bad.canonical_name)",+      "      FOR UPDATE;",+      "",+      "    table_exists := EXISTS (",+      "      SELECT 1",+      "      FROM pg_catalog.pg_class c",+      "      JOIN pg_catalog.pg_namespace n ON n.oid = c.relnamespace",+      "      WHERE n.nspname = 'pgmq' AND c.relname = 'q_' || bad.canonical_name",+      "    );",+      "    IF NOT table_exists THEN",+      "      DELETE FROM pgmq.meta WHERE queue_name = bad.mixed_name;",+      "      CONTINUE;",+      "    END IF;",+      "",+      "    twin_exists := EXISTS (",+      "      SELECT 1 FROM pgmq.meta WHERE queue_name = bad.canonical_name",+      "    );",+      "",+      "    IF NOT twin_exists THEN",+      "      INSERT INTO pgmq.meta (queue_name, is_partitioned, is_unlogged, created_at)",+      "      SELECT bad.canonical_name, m.is_partitioned, m.is_unlogged, m.created_at",+      "      FROM pgmq.meta m WHERE m.queue_name = bad.mixed_name;",+      "    ELSE",+      "      DELETE FROM pgmq.topic_bindings b",+      "      WHERE b.queue_name = bad.mixed_name",+      "        AND EXISTS (",+      "          SELECT 1 FROM pgmq.topic_bindings t",+      "          WHERE t.queue_name = bad.canonical_name AND t.pattern = b.pattern",+      "        );",+      "      DELETE FROM pgmq.notify_insert_throttle",+      "      WHERE queue_name = bad.mixed_name",+      "        AND EXISTS (",+      "          SELECT 1 FROM pgmq.notify_insert_throttle",+      "          WHERE queue_name = bad.canonical_name",+      "        );",+      "    END IF;",+      "",+      "    UPDATE pgmq.topic_bindings SET queue_name = bad.canonical_name",+      "    WHERE queue_name = bad.mixed_name;",+      "    UPDATE pgmq.notify_insert_throttle SET queue_name = bad.canonical_name",+      "    WHERE queue_name = bad.mixed_name;",+      "",+      "    DELETE FROM pgmq.meta WHERE queue_name = bad.mixed_name;",+      "  END LOOP;",+      "END",+      "$remediate$"+    ]++-- | Detection query from design note 016, reduced to a count.+detectionCount :: Session Int64+detectionCount = rawCount "select count(*) from pgmq.meta m where m.queue_name <> lower(m.queue_name)"++-- | No lowercase twin: the mixed-case row is renamed in place; both topic+-- bindings (with their @bound_at@) and the throttle configuration survive+-- under the canonical name, and the notification trigger starts matching.+testNoTwinRename :: IO (Pg.Database, Pool.Pool) -> TestTree+testNoTwinRename getDb = testCase "no-twin rename preserves bindings and throttle configuration" $ do+  (_, pool) <- getDb+  suffix <- genSuffix+  let mixed = "Legacy_" <> suffix+      canonical = "legacy_" <> suffix+  assertSession pool (rawUnit ("select pgmq.create('" <> mixed <> "')"))+  assertSession pool (rawUnit ("select pgmq.enable_notify_insert('" <> mixed <> "', 750)"))+  assertSession pool (rawUnit ("select pgmq.bind_topic('orders.*', '" <> mixed <> "')"))+  assertSession pool (rawUnit ("select pgmq.bind_topic('audit.#', '" <> mixed <> "')"))+  bindingsBefore <- assertSession pool (bindingFingerprints mixed)+  assertEqual "Seeded two bindings on the mixed-case row" 2 (V.length bindingsBefore)++  assertSession pool (rawUnit remediationSql)++  detected <- assertSession pool detectionCount+  assertEqual "Detection query finds nothing after remediation" 0 detected+  canonicalMeta <- assertSession pool (rawCount ("select count(*) from pgmq.meta where queue_name = '" <> canonical <> "'"))+  assertEqual "The canonical meta row exists" 1 canonicalMeta+  mixedMeta <- assertSession pool (rawCount ("select count(*) from pgmq.meta where queue_name = '" <> mixed <> "'"))+  assertEqual "The mixed-case meta row is gone" 0 mixedMeta+  bindingsAfter <- assertSession pool (bindingFingerprints canonical)+  assertEqual+    "Both bindings survive under the canonical name with bound_at preserved"+    (V.toList bindingsBefore)+    (V.toList bindingsAfter)+  interval <- assertSession pool (throttleInterval canonical)+  assertEqual "The throttle configuration survives under the canonical name" 750 interval++  -- Functional proof: the trigger's lowercase lookup now matches the throttle+  -- row, so a send stamps last_notified_at off the epoch.+  void $ assertSession pool (rawIds ("select pgmq.send('" <> canonical <> "', '{\"probe\":true}'::jsonb)"))+  stamped <-+    assertSession pool $+      rawBool+        ( "select last_notified_at > to_timestamp(0) from pgmq.notify_insert_throttle where queue_name = '"+            <> canonical+            <> "'"+        )+  assertBool "After remediation the trigger matches and stamps the throttle row" stamped++  -- Rerun: the remediation must be a no-op now.+  assertSession pool (rawUnit remediationSql)+  bindingsRerun <- assertSession pool (bindingFingerprints canonical)+  assertEqual "A second run changes no bindings" (V.toList bindingsAfter) (V.toList bindingsRerun)+  intervalRerun <- assertSession pool (throttleInterval canonical)+  assertEqual "A second run changes no throttle configuration" 750 intervalRerun+  void $ assertSession pool (rawBool ("select pgmq.drop_queue('" <> canonical <> "')"))++-- | A lowercase twin exists: the two rows already alias one physical table.+-- Bindings move to the twin (duplicates deduplicate), the twin's own throttle+-- configuration wins, and the mixed-case row disappears.+testTwinMerge :: IO (Pg.Database, Pool.Pool) -> TestTree+testTwinMerge getDb = testCase "twin merge moves bindings, dedupes, and keeps the canonical throttle" $ do+  (_, pool) <- getDb+  suffix <- genSuffix+  let mixed = "Shared_" <> suffix+      canonical = "shared_" <> suffix+  assertSession pool (rawUnit ("select pgmq.create('" <> mixed <> "')"))+  assertSession pool (rawUnit ("select pgmq.create('" <> canonical <> "')"))+  assertSession pool (rawUnit ("select pgmq.bind_topic('dup.*', '" <> mixed <> "')"))+  assertSession pool (rawUnit ("select pgmq.bind_topic('dup.*', '" <> canonical <> "')"))+  assertSession pool (rawUnit ("select pgmq.bind_topic('only.*', '" <> mixed <> "')"))+  assertSession pool (rawUnit ("select pgmq.enable_notify_insert('" <> mixed <> "', 900)"))+  assertSession pool (rawUnit ("select pgmq.enable_notify_insert('" <> canonical <> "', 250)"))++  assertSession pool (rawUnit remediationSql)++  detected <- assertSession pool detectionCount+  assertEqual "Detection query finds nothing after remediation" 0 detected+  metaRows <- assertSession pool (rawCount ("select count(*) from pgmq.meta where lower(queue_name) = '" <> canonical <> "'"))+  assertEqual "One meta row remains for the pair" 1 metaRows+  patterns <- assertSession pool (bindingPatterns canonical)+  assertEqual+    "The twin holds the union of bindings, duplicates collapsed"+    ["dup.*", "only.*"]+    (V.toList patterns)+  orphanBindings <- assertSession pool (rawCount ("select count(*) from pgmq.topic_bindings where queue_name = '" <> mixed <> "'"))+  assertEqual "No bindings remain under the mixed-case name" 0 orphanBindings+  interval <- assertSession pool (throttleInterval canonical)+  assertEqual "The canonical queue's own throttle configuration wins" 250 interval+  orphanThrottles <- assertSession pool (rawCount ("select count(*) from pgmq.notify_insert_throttle where queue_name = '" <> mixed <> "'"))+  assertEqual "No throttle row remains under the mixed-case name" 0 orphanThrottles++  -- Rerun: still nothing to do.+  assertSession pool (rawUnit remediationSql)+  patternsRerun <- assertSession pool (bindingPatterns canonical)+  assertEqual "A second run changes no bindings" (V.toList patterns) (V.toList patternsRerun)+  void $ assertSession pool (rawBool ("select pgmq.drop_queue('" <> canonical <> "')"))++-- | The physical table is gone: @drop_queue@ on the lowercase twin destroyed+-- the shared table and deleted its own meta row, leaving the mixed-case row+-- pointing at nothing (AliasingSpec demonstrates the state live).+-- Canonicalizing that orphan would insert a meta row for a queue with no+-- table — it lists cleanly and fails every send with 42P01 — so the+-- remediation must delete it, cascading away children that route to nothing.+testOrphanDeletion :: IO (Pg.Database, Pool.Pool) -> TestTree+testOrphanDeletion getDb = testCase "orphaned row is deleted, not resurrected as a phantom queue" $ do+  (_, pool) <- getDb+  suffix <- genSuffix+  let mixed = "Ghost_" <> suffix+      canonical = "ghost_" <> suffix+  assertSession pool (rawUnit ("select pgmq.create('" <> mixed <> "')"))+  assertSession pool (rawUnit ("select pgmq.create('" <> canonical <> "')"))+  assertSession pool (rawUnit ("select pgmq.bind_topic('ghost.*', '" <> mixed <> "')"))+  assertSession pool (rawUnit ("select pgmq.enable_notify_insert('" <> mixed <> "', 500)"))+  void $ assertSession pool (rawBool ("select pgmq.drop_queue('" <> canonical <> "')"))++  -- Premise: the shared physical table is destroyed, the mixed-case row and+  -- its children survive it.+  tableCount <- assertSession pool (physicalTableCount canonical)+  assertEqual "The shared physical table is gone" 0 tableCount+  mixedBefore <- assertSession pool (rawCount ("select count(*) from pgmq.meta where queue_name = '" <> mixed <> "'"))+  assertEqual "The mixed-case meta row is orphaned, not dropped" 1 mixedBefore+  bindingsBefore <- assertSession pool (rawCount ("select count(*) from pgmq.topic_bindings where queue_name = '" <> mixed <> "'"))+  assertEqual "The orphan still carries its binding" 1 bindingsBefore+  throttleBefore <- assertSession pool (rawCount ("select count(*) from pgmq.notify_insert_throttle where queue_name = '" <> mixed <> "'"))+  assertEqual "The orphan still carries its throttle row" 1 throttleBefore++  assertSession pool (rawUnit remediationSql)++  detected <- assertSession pool detectionCount+  assertEqual "Detection query finds nothing after remediation" 0 detected+  metaAfter <- assertSession pool (rawCount ("select count(*) from pgmq.meta where lower(queue_name) = '" <> canonical <> "'"))+  assertEqual "No meta row remains under either casing — no phantom queue" 0 metaAfter+  bindingsAfter <- assertSession pool (rawCount ("select count(*) from pgmq.topic_bindings where lower(queue_name) = '" <> canonical <> "'"))+  assertEqual "The CASCADE removed the orphan's binding" 0 bindingsAfter+  throttleAfter <- assertSession pool (rawCount ("select count(*) from pgmq.notify_insert_throttle where lower(queue_name) = '" <> canonical <> "'"))+  assertEqual "The CASCADE removed the orphan's throttle row" 0 throttleAfter++  -- Rerun: still nothing to do.+  assertSession pool (rawUnit remediationSql)+  detectedRerun <- assertSession pool detectionCount+  assertEqual "A second run still finds nothing" 0 detectedRerun++-- Dedicated database plumbing -------------------------------------------------++acquireDb :: IO (Pg.Database, Pool.Pool)+acquireDb = do+  started <- Pg.startCached Pg.defaultConfig Pg.defaultCacheConfig+  db <- either (\err -> error ("could not start a dedicated PostgreSQL: " <> show err)) pure started+  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 ()+  pool <-+    Pool.acquire $+      PoolConfig.settings+        [ PoolConfig.size 2,+          PoolConfig.staticConnectionSettings (Pg.connectionSettings db)+        ]+  pure (db, pool)++releaseDb :: (Pg.Database, Pool.Pool) -> IO ()+releaseDb (db, pool) = do+  Pool.release pool+  Pg.stop db++-- Raw statement helpers -------------------------------------------------------+--+-- Queue names are spliced into the SQL text because these tests must construct+-- names the Haskell API (rightly) refuses; every spliced value is generated+-- below from @[A-Za-z0-9_]@.++rawUnit :: Text -> Session ()+rawUnit sqlText = statement () (unpreparable sqlText mempty D.noResult)++rawCount :: Text -> Session Int64+rawCount sqlText = statement () (unpreparable sqlText mempty (D.singleRow (D.column (D.nonNullable D.int8))))++rawBool :: Text -> Session Bool+rawBool sqlText = statement () (unpreparable sqlText mempty (D.singleRow (D.column (D.nonNullable D.bool))))++rawIds :: Text -> Session (V.Vector Int64)+rawIds sqlText = statement () (unpreparable sqlText mempty (D.rowVector (D.column (D.nonNullable D.int8))))++rawTexts :: Text -> Session (V.Vector Text)+rawTexts sqlText = statement () (unpreparable sqlText mempty (D.rowVector (D.column (D.nonNullable D.text))))++-- | Pattern plus creation timestamp, so equality across the remediation proves+-- @bound_at@ survived, not merely the pattern.+bindingFingerprints :: Text -> Session (V.Vector Text)+bindingFingerprints qname =+  rawTexts+    ( "select pattern || '|' || bound_at::text from pgmq.topic_bindings where queue_name = '"+        <> qname+        <> "' order by pattern"+    )++bindingPatterns :: Text -> Session (V.Vector Text)+bindingPatterns qname =+  rawTexts ("select pattern from pgmq.topic_bindings where queue_name = '" <> qname <> "' order by pattern")++throttleInterval :: Text -> Session Int64+throttleInterval qname =+  rawCount ("select throttle_interval_ms::int8 from pgmq.notify_insert_throttle where queue_name = '" <> qname <> "'")++-- | Same probe the remediation's orphan branch uses.+physicalTableCount :: Text -> Session Int64+physicalTableCount canonical =+  rawCount+    ( "select count(*) from pg_catalog.pg_class c"+        <> " join pg_catalog.pg_namespace n on n.oid = c.relnamespace"+        <> " where n.nspname = 'pgmq' and c.relname = 'q_"+        <> canonical+        <> "'"+    )++assertSession :: Pool.Pool -> Session a -> IO a+assertSession pool session = do+  result <- Pool.use pool session+  case result of+    Left err -> assertFailure $ "Session failed: " <> show err+    Right a -> pure a++genSuffix :: IO Text+genSuffix = do+  suffix <- randomRIO (10000 :: Word32, 99999)+  pure (T.pack (show suffix))

@@ -0,0 +1,129 @@+{-# LANGUAGE OverloadedStrings #-}++-- | PGH-9: the documented LISTEN\/NOTIFY channel name was wrong everywhere.+--+-- The Haddock and the design note both claimed @pgmq_\<queue_name\>@, so anyone+-- following them listened on a channel that never receives anything. The real+-- channel is @pgmq.q_\<lowercased queue name\>.INSERT@, now computed by+-- 'notifyChannelName'.+--+-- This module pins the contract from both sides: a real notification arrives on+-- exactly the channel the helper computes, and a listener on the old documented+-- name receives nothing.+module NotifyChannelSpec (tests) where++import Control.Concurrent (threadDelay)+import Control.Exception (bracket)+import Control.Monad (unless)+import Data.Aeson qualified as Aeson+import Data.Text (Text)+import Data.Text qualified as T+import Data.Text.Encoding qualified as TE+import Database.PostgreSQL.LibPQ qualified as LibPQ+import EphemeralDb (Database, TestFixture (..), withTestFixture)+import EphemeralPg qualified as Pg+import Hasql.Pool qualified as Pool+import Pgmq.Hasql.Sessions qualified as Sessions+import Pgmq.Hasql.Statements.Types qualified as StmtTypes+import Pgmq.Types (MessageBody (..), QueueName, notifyChannelName, queueNameToText)+import Test.Tasty (TestTree, testGroup)+import Test.Tasty.HUnit (assertFailure, testCase, (@?=))+import TestUtils (assertSession, cleanupQueue)++tests :: Pool.Pool -> Database -> TestTree+tests p db =+  testGroup+    "Notification Channel Contract"+    [ testCase "a notification arrives on exactly notifyChannelName" $+        withTestFixture p $ \TestFixture {pool, queueName} -> do+          enableNotify pool queueName+          received <- withListener db (notifyChannelName queueName) $ \conn -> do+            sendProbe pool queueName+            awaitNotify conn 20+          case received of+            Nothing ->+              assertFailure $+                "expected a notification on " <> show (notifyChannelName queueName) <> " within 2s, got none"+            Just notification ->+              LibPQ.notifyRelname notification @?= TE.encodeUtf8 (notifyChannelName queueName)+          cleanupQueue pool queueName,+      testCase "nothing arrives on the old documented channel name" $+        withTestFixture p $ \TestFixture {pool, queueName} -> do+          enableNotify pool queueName+          received <- withListener db (legacyChannelName queueName) $ \conn -> do+            sendProbe pool queueName+            awaitNotify conn 10+          case received of+            Nothing -> pure ()+            Just notification ->+              assertFailure $+                "the old documented channel "+                  <> show (legacyChannelName queueName)+                  <> " received "+                  <> show (LibPQ.notifyRelname notification)+          cleanupQueue pool queueName+    ]++-- | The channel name this library's documentation claimed until 2026-08-05.+legacyChannelName :: QueueName -> Text+legacyChannelName qn = "pgmq_" <> queueNameToText qn++enableNotify :: Pool.Pool -> QueueName -> IO ()+enableNotify pool qn = do+  assertSession pool (Sessions.createQueue qn)+  assertSession pool $+    Sessions.enableNotifyInsert+      StmtTypes.EnableNotifyInsert+        { StmtTypes.queueName = qn,+          StmtTypes.throttleIntervalMs = Just 0 -- 0 = never throttle+        }++sendProbe :: Pool.Pool -> QueueName -> IO ()+sendProbe pool qn =+  ()+    <$ assertSession+      pool+      ( Sessions.sendMessage+          StmtTypes.SendMessage+            { StmtTypes.queueName = qn,+              StmtTypes.messageBody = MessageBody (Aeson.String "notify-probe"),+              StmtTypes.delay = Nothing+            }+      )++-- | Open a raw libpq connection (hasql 1.10 exposes no notification API) and+-- subscribe to @channel@. ephemeral-pg hands out connection strings as 'Text'+-- while libpq consumes 'ByteString', so both the conninfo and the command are+-- encoded explicitly.+withListener :: 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)

@@ -0,0 +1,103 @@+{-# LANGUAGE OverloadedStrings #-}++-- | PGH-8: two replicas reconciling the same configuration concurrently must not+-- collide on notification setup.+--+-- @pgmq.enable_notify_insert@ takes no lock and its @CREATE CONSTRAINT TRIGGER@+-- has no @IF NOT EXISTS@. Two callers can both pass the internal+-- @DROP TRIGGER IF EXISTS@ (a no-op on a fresh queue, so neither locks the+-- table); the second then blocks on the throttle-row unique constraint until the+-- first commits, resumes, and creates a trigger that now already exists —+-- SQLSTATE 42710, failing that replica's whole startup reconcile.+module NotifyRaceSpec (tests) where++import Control.Concurrent (forkIO)+import Control.Concurrent.MVar (newEmptyMVar, putMVar, readMVar, takeMVar)+import Control.Exception (SomeException, try)+import Data.Text qualified as T+import Data.Word (Word32)+import Hasql.Pool qualified as Pool+import Hasql.Session (Session)+import Pgmq.Hasql.Sessions qualified as Sessions+import Pgmq.Hasql.Statements.Types qualified as StmtTypes+import Pgmq.Types (QueueName, parseQueueName)+import System.Random (randomRIO)+import Test.Tasty (TestTree, testGroup)+import Test.Tasty.HUnit (assertFailure, testCase)++-- | How many fresh queues to race over. The losing caller only errors when it+-- passes the internal DROP before the winner commits its CREATE, which is a+-- narrow window — measured at roughly one collision per hundred calls — so the+-- count is deliberately generous.+iterations :: Int+iterations = 200++tests :: Pool.Pool -> TestTree+tests pool =+  testGroup+    "Notification Enable Race"+    [ testCase ("concurrent enable_notify_insert never raises 42710 (n=" <> show iterations <> ")") $ do+        failures <- concat <$> traverse (const (raceOnce pool)) [1 .. iterations]+        case filter isDuplicateObject failures of+          [] -> pure ()+          duplicates@(firstDuplicate : _) ->+            assertFailure $+              show (length duplicates)+                <> " of "+                <> show (2 * iterations)+                <> " concurrent enable_notify_insert calls failed with duplicate_object (42710). First: "+                <> firstDuplicate+    ]++-- | Create a fresh queue, enable notify on it from two connections at once, and+-- return whatever went wrong.+raceOnce :: Pool.Pool -> IO [String]+raceOnce pool = do+  qn <- genQueueName+  created <- runSession pool (Sessions.createQueue qn)+  case created of+    Left err -> pure [err]+    Right () -> do+      gate <- newEmptyMVar+      leftSlot <- newEmptyMVar+      rightSlot <- newEmptyMVar+      let enable slot = do+            () <- readMVar gate+            result <- runSession pool (enableSession qn)+            putMVar slot result+      _ <- forkIO (enable leftSlot)+      _ <- forkIO (enable rightSlot)+      putMVar gate ()+      leftResult <- takeMVar leftSlot+      rightResult <- takeMVar rightSlot+      _ <- runSession pool (() <$ Sessions.dropQueue qn)+      pure [err | Left err <- [leftResult, rightResult]]++enableSession :: QueueName -> Session ()+enableSession qn =+  Sessions.enableNotifyInsert+    StmtTypes.EnableNotifyInsert+      { StmtTypes.queueName = qn,+        StmtTypes.throttleIntervalMs = Just 0+      }++-- | Run a session, flattening both pool errors and thrown exceptions into a+-- printable failure so a forked thread can never leave its 'MVar' empty.+runSession :: Pool.Pool -> Session a -> IO (Either String a)+runSession pool session = do+  outcome <- try (Pool.use pool session)+  pure $ case outcome of+    Left (e :: SomeException) -> Left (show e)+    Right (Left usageError) -> Left (show usageError)+    Right (Right a) -> Right a++-- | hasql renders the SQLSTATE into the shown 'Pool.UsageError'.+isDuplicateObject :: String -> Bool+isDuplicateObject = T.isInfixOf "42710" . T.pack++genQueueName :: IO QueueName+genQueueName = do+  suffix <- randomRIO (10000 :: Word32, 99999)+  case parseQueueName ("race_test_" <> T.pack (show suffix)) of+    Left err -> error $ "Failed to generate queue name: " <> show err+    Right qn -> pure qn

@@ -0,0 +1,158 @@+{-# LANGUAGE OverloadedStrings #-}++-- | PGH-11: a SQL NULL message body must not poison every read batch.+--+-- The queue table's @message@ column is nullable and+-- @select pgmq.send('q', null::jsonb)@ is legal SQL, so any non-Haskell+-- producer (psql, another language's client, a trigger) can insert a NULL+-- body. The decoder required a non-null body, so every batch containing such a+-- row failed at decode — /after/ the read statement had already bumped @vt@ and+-- @read_ct@ for the whole batch, because the statement succeeded and only its+-- result failed to decode. The row could not be seen, read, or archived through+-- the Haskell client, and it re-poisoned every batch each time its visibility+-- timeout lapsed.+--+-- The fix decodes SQL NULL as JSON @null@ (@MessageBody Aeson.Null@), an+-- accepted conflation with an explicitly-sent JSON @null@ body: both mean "no+-- usable payload", and the poison row becomes visible, identifiable, and+-- archivable through the normal API.+module NullBodySpec (tests) where++import Data.Aeson (object, (.=))+import Data.Aeson qualified as Aeson+import Data.Int (Int64)+import Data.Text (Text)+import Data.Vector qualified as V+import EphemeralDb (TestFixture (..), withTestFixture)+import Hasql.Decoders qualified as D+import Hasql.Pool qualified as Pool+import Hasql.Session (Session, statement)+import Hasql.Statement (unpreparable)+import Pgmq.Hasql.Sessions qualified as Sessions+import Pgmq.Hasql.Statements.Types+  ( BatchSendMessage (..),+    MessageQuery (..),+    ReadMessage (..),+  )+import Pgmq.Types (MessageBody (..), queueNameToText)+import Pgmq.Types qualified as PgmqTypes+import Test.Tasty (TestTree, testGroup)+import Test.Tasty.HUnit (assertBool, assertEqual, assertFailure, testCase)+import TestUtils (assertSession, cleanupQueue, runSession)++tests :: Pool.Pool -> TestTree+tests p =+  testGroup+    "NULL Message Body (PGH-11)"+    [ testNullBodyBatchReadsFully p,+      testNullBodyArchivable p,+      testReadCtBumpedRegardless p+    ]++-- | Seed two well-formed messages through the API, then insert the poison row+-- the way any non-Haskell producer would: raw SQL. The queue name comes from+-- the fixture generator, so splicing it into the SQL text is safe.+seedWithPoison :: Pool.Pool -> PgmqTypes.QueueName -> IO ()+seedWithPoison pool queueName = do+  assertSession pool (Sessions.createQueue queueName)+  _ <-+    assertSession pool $+      Sessions.batchSendMessage+        BatchSendMessage+          { queueName = queueName,+            messageBodies = [MessageBody (object ["seq" .= i]) | i <- [1 :: Int, 2]],+            delay = Nothing+          }+  _ <-+    assertSession pool $+      rawIds ("select pgmq.send('" <> queueNameToText queueName <> "', null::jsonb)")+  pure ()++-- | A batch containing the NULL-bodied row must read fully, with the poison+-- row surfacing as JSON @null@. Red before the decoder fix: the whole batch+-- failed with a decode error on the NULL cell.+testNullBodyBatchReadsFully :: Pool.Pool -> TestTree+testNullBodyBatchReadsFully p = testCase "a batch containing a NULL body reads fully" $ do+  withTestFixture p $ \TestFixture {pool, queueName} -> do+    seedWithPoison pool queueName+    msgs <-+      assertSession pool $+        Sessions.readMessage+          ReadMessage+            { queueName = queueName,+              delay = 30,+              batchSize = Just 10,+              conditional = Nothing+            }+    assertEqual "All three messages read, poison row included" 3 (V.length msgs)+    let nullBodied = [m | m <- V.toList msgs, unMessageBody (PgmqTypes.body m) == Aeson.Null]+    assertEqual "Exactly one message surfaces as JSON null" 1 (length nullBodied)+    cleanupQueue pool queueName++-- | The poison row must be identifiable and archivable through the normal API+-- — the dead-letter path a consumer actually needs. Red before the fix: the+-- row could not even be read to learn its id.+testNullBodyArchivable :: Pool.Pool -> TestTree+testNullBodyArchivable p = testCase "the NULL-bodied row can be archived through the normal API" $ do+  withTestFixture p $ \TestFixture {pool, queueName} -> do+    seedWithPoison pool queueName+    msgs <-+      assertSession pool $+        Sessions.readMessage+          ReadMessage+            { queueName = queueName,+              delay = 30,+              batchSize = Just 10,+              conditional = Nothing+            }+    poisonId <-+      case [PgmqTypes.messageId m | m <- V.toList msgs, unMessageBody (PgmqTypes.body m) == Aeson.Null] of+        [msgId] -> pure msgId+        other -> assertFailure $ "Expected exactly one NULL-bodied message, got " <> show (length other)+    archived <-+      assertSession pool $+        Sessions.archiveMessage MessageQuery {queueName = queueName, messageId = poisonId}+    assertBool "archiveMessage reports success for the poison row" archived+    remaining <- assertSession pool (rawCount ("select count(*) from pgmq.q_" <> queueNameToText queueName))+    assertEqual "The two well-formed messages remain queued" 2 remaining+    archivedCount <- assertSession pool (rawCount ("select count(*) from pgmq.a_" <> queueNameToText queueName))+    assertEqual "The poison row landed in the archive" 1 archivedCount+    cleanupQueue pool queueName++-- | The read statement bumps @read_ct@ and @vt@ for the whole batch whether or+-- not the client manages to decode the result. Before the decoder fix this is+-- what made the NULL body a poison row rather than a mere error: the failed+-- call still consumed a read attempt for every batch-mate and hid the whole+-- batch for the visibility timeout, over and over. This test passes before and+-- after the fix; before, it documents the damage the failed call left behind.+testReadCtBumpedRegardless :: Pool.Pool -> TestTree+testReadCtBumpedRegardless p = testCase "read_ct is bumped for the whole batch even when decode fails" $ do+  withTestFixture p $ \TestFixture {pool, queueName} -> do+    seedWithPoison pool queueName+    -- Deliberately ignore the outcome: Left (decode failure) before the fix,+    -- Right afterwards. The server-side damage is identical.+    _ <-+      runSession pool $+        Sessions.readMessage+          ReadMessage+            { queueName = queueName,+              delay = 30,+              batchSize = Just 10,+              conditional = Nothing+            }+    readCts <-+      assertSession pool $+        rawCounts ("select read_ct::int8 from pgmq.q_" <> queueNameToText queueName <> " order by msg_id")+    assertEqual "All three rows consumed a read attempt" [1, 1, 1] (V.toList readCts)+    cleanupQueue pool queueName++-- Raw statement helpers -------------------------------------------------------++rawIds :: Text -> Session (V.Vector Int64)+rawIds sqlText = statement () (unpreparable sqlText mempty (D.rowVector (D.column (D.nonNullable D.int8))))++rawCount :: Text -> Session Int64+rawCount sqlText = statement () (unpreparable sqlText mempty (D.singleRow (D.column (D.nonNullable D.int8))))++rawCounts :: Text -> Session (V.Vector Int64)+rawCounts sqlText = statement () (unpreparable sqlText mempty (D.rowVector (D.column (D.nonNullable D.int8))))

@@ -0,0 +1,300 @@+{-# LANGUAGE OverloadedStrings #-}++-- | Tests that pin the meaning of @Nothing@ for every optional parameter that+-- reaches PostgreSQL as a bound SQL NULL.+--+-- Background a reader needs: a plpgsql parameter DEFAULT applies only when the+-- argument is /omitted/ from the call. A bound SQL NULL is a supplied argument,+-- so it silently overrides the DEFAULT. Combined with @LIMIT NULL@ meaning+-- @LIMIT ALL@ in PostgreSQL, an optional batch size encoded as a nullable+-- parameter turns "no preference" into "the whole queue". These tests assert+-- the documented behaviour instead: @Nothing@ means the documented default and+-- never widens the scope of an operation.+module NullSemanticsSpec (tests) where++import Data.Aeson (object, (.=))+import Data.Text (Text)+import Data.Time.Clock (addUTCTime, getCurrentTime)+import Data.Vector qualified as V+import EphemeralDb (TestFixture (..), withTestFixture)+import Hasql.Pool qualified as Pool+import Pgmq.Hasql.Sessions qualified as Sessions+import Pgmq.Hasql.Statements.Types+  ( BatchSendMessage (..),+    EnableNotifyInsert (..),+    PopMessage (..),+    QueueMetrics (..),+    ReadMessage (..),+    ReadWithPollMessage (..),+    SendMessage (..),+    VisibilityTimeoutAtQuery (..),+    VisibilityTimeoutQuery (..),+  )+import Pgmq.Types (MessageBody (..), MessageId (..), queueNameToText)+import Pgmq.Types qualified as PgmqTypes+import Test.Tasty (TestTree, testGroup)+import Test.Tasty.HUnit (assertEqual, assertFailure, testCase)+import TestUtils (assertJust, assertSession, cleanupQueue)++-- | All NULL-parameter semantics tests+tests :: Pool.Pool -> TestTree+tests p =+  testGroup+    "NULL Parameter Semantics"+    [ testPopNothingDoesNotDrainQueue p,+      testReadNothingDoesNotLeaseQueue p,+      testReadWithPollNothingDoesNotLeaseQueue p,+      testConditionalFiltersWhenJust p,+      testEnableNotifyInsertNothingUsesDefault p,+      testSetVtOnMissingRow p+    ]++-- | Send @n@ distinct messages to a queue and return nothing useful; the bodies+-- are irrelevant beyond being distinguishable in a failure message.+seedMessages :: Pool.Pool -> PgmqTypes.QueueName -> Int -> IO ()+seedMessages pool queueName n = do+  _ <-+    assertSession pool $+      Sessions.batchSendMessage+        BatchSendMessage+          { queueName = queueName,+            messageBodies = [MessageBody (object ["seq" .= i]) | i <- [1 .. n]],+            delay = Nothing+          }+  pure ()++-- | @pop@ with @qty = Nothing@ must pop exactly one message.+--+-- Before the fix this popped — and therefore permanently deleted — every+-- visible message in the queue, because the NULL @qty@ became @LIMIT ALL@ in+-- the DELETE-returning CTE inside @pgmq.pop@. There is no visibility-timeout+-- safety net for @pop@: the rows are gone.+testPopNothingDoesNotDrainQueue :: Pool.Pool -> TestTree+testPopNothingDoesNotDrainQueue p = testCase "pop with qty = Nothing pops exactly one message" $ do+  withTestFixture p $ \TestFixture {pool, queueName} -> do+    assertSession pool (Sessions.createQueue queueName)+    seedMessages pool queueName 5+    popped <- assertSession pool (Sessions.pop PopMessage {queueName = queueName, qty = Nothing})+    assertEqual "Should pop exactly 1 message" 1 (V.length popped)+    metrics <- assertSession pool (Sessions.queueMetrics queueName)+    assertEqual "Should leave 4 messages in the queue" 4 (queueLength metrics)+    cleanupQueue pool queueName++-- | @read@ with @batchSize = Nothing@ must read exactly one message.+--+-- Before the fix the NULL batch size became @LIMIT ALL@, so a single call+-- leased the entire queue: every row had its visibility timeout pushed forward+-- and its read count incremented, hiding the whole queue from every other+-- consumer for the duration of the timeout.+testReadNothingDoesNotLeaseQueue :: Pool.Pool -> TestTree+testReadNothingDoesNotLeaseQueue p = testCase "read with batchSize = Nothing reads exactly one message" $ do+  withTestFixture p $ \TestFixture {pool, queueName} -> do+    assertSession pool (Sessions.createQueue queueName)+    seedMessages pool queueName 5+    first <-+      assertSession pool $+        Sessions.readMessage+          ReadMessage+            { queueName = queueName,+              delay = 30,+              batchSize = Nothing,+              conditional = Nothing+            }+    assertEqual "Should read exactly 1 message" 1 (V.length first)+    -- Only one row may have been leased, so four remain immediately visible.+    second <-+      assertSession pool $+        Sessions.readMessage+          ReadMessage+            { queueName = queueName,+              delay = 30,+              batchSize = Just 10,+              conditional = Nothing+            }+    assertEqual "Should leave 4 messages unleased" 4 (V.length second)+    cleanupQueue pool queueName++-- | @readWithPoll@ shares the @LIMIT NULL@ hazard with @read@ and must behave+-- identically for @batchSize = Nothing@.+testReadWithPollNothingDoesNotLeaseQueue :: Pool.Pool -> TestTree+testReadWithPollNothingDoesNotLeaseQueue p = testCase "readWithPoll with batchSize = Nothing reads exactly one message" $ do+  withTestFixture p $ \TestFixture {pool, queueName} -> do+    assertSession pool (Sessions.createQueue queueName)+    seedMessages pool queueName 5+    first <-+      assertSession pool $+        Sessions.readWithPoll+          ReadWithPollMessage+            { queueName = queueName,+              delay = 30,+              batchSize = Nothing,+              maxPollSeconds = 1,+              pollIntervalMs = 100,+              conditional = Nothing+            }+    assertEqual "Should read exactly 1 message" 1 (V.length first)+    second <-+      assertSession pool $+        Sessions.readWithPoll+          ReadWithPollMessage+            { queueName = queueName,+              delay = 30,+              batchSize = Just 10,+              maxPollSeconds = 1,+              pollIntervalMs = 100,+              conditional = Nothing+            }+    assertEqual "Should leave 4 messages unleased" 4 (V.length second)+    cleanupQueue pool queueName++-- | The @conditional@ field on 'ReadMessage' must actually filter.+--+-- @conditional@ is a JSONB containment filter: a message is returned only when+-- its body contains the given object (SQL @message \@> conditional@).+-- @Nothing@ means "no filtering".+--+-- Before the fix the field was never encoded — the statement bound only three+-- parameters — so a @Just@ filter was silently ignored and every visible+-- message came back.+testConditionalFiltersWhenJust :: Pool.Pool -> TestTree+testConditionalFiltersWhenJust p = testCase "conditional filters when Just and is neutral when Nothing" $ do+  withTestFixture p $ \TestFixture {pool, queueName} -> do+    assertSession pool (Sessions.createQueue queueName)+    _ <-+      assertSession pool $+        Sessions.batchSendMessage+          BatchSendMessage+            { queueName = queueName,+              messageBodies =+                [ MessageBody (object ["kind" .= ("a" :: Text)]),+                  MessageBody (object ["kind" .= ("b" :: Text)])+                ],+              delay = Nothing+            }+    -- delay = 0 keeps both messages immediately visible for the second read.+    filtered <-+      assertSession pool $+        Sessions.readMessage+          ReadMessage+            { queueName = queueName,+              delay = 0,+              batchSize = Just 10,+              conditional = Just (object ["kind" .= ("a" :: Text)])+            }+    assertEqual "Filtered read should return only the matching message" 1 (V.length filtered)+    case V.toList filtered of+      [msg] ->+        assertEqual+          "Filtered read should return the 'a' message"+          (object ["kind" .= ("a" :: Text)])+          (unMessageBody (PgmqTypes.body msg))+      _ -> assertFailure "Filtered read should return exactly one message"+    unfiltered <-+      assertSession pool $+        Sessions.readMessage+          ReadMessage+            { queueName = queueName,+              delay = 0,+              batchSize = Just 10,+              conditional = Nothing+            }+    assertEqual "Unfiltered read should return both messages" 2 (V.length unfiltered)+    cleanupQueue pool queueName++-- | @enableNotifyInsert@ with @throttleIntervalMs = Nothing@ must install the+-- documented 250 ms throttle.+--+-- Before the fix the bound SQL NULL was inserted straight into+-- @pgmq.notify_insert_throttle.throttle_interval_ms@, which is @NOT NULL@; a+-- column DEFAULT does not apply to an explicitly supplied NULL, so the call+-- raised SQLSTATE 23502 every single time. Because the reconciler runs each+-- statement in its own transaction, the queue creation had already committed,+-- so the failure repeated on every application startup forever.+testEnableNotifyInsertNothingUsesDefault :: Pool.Pool -> TestTree+testEnableNotifyInsertNothingUsesDefault p = testCase "enableNotifyInsert with Nothing applies the 250ms default" $ do+  withTestFixture p $ \TestFixture {pool, queueName} -> do+    assertSession pool (Sessions.createQueue queueName)+    assertSession pool $+      Sessions.enableNotifyInsert+        EnableNotifyInsert {queueName = queueName, throttleIntervalMs = Nothing}+    throttles <- assertSession pool Sessions.listNotifyInsertThrottles+    let mine = filter (\t -> PgmqTypes.throttleQueueName t == queueNameToText queueName) throttles+    case mine of+      [t] ->+        assertEqual+          "Throttle interval should be the documented 250ms default"+          250+          (PgmqTypes.throttleIntervalMs t)+      _ -> assertFailure $ "Expected exactly one throttle row for the queue, got " <> show (length mine)+    cleanupQueue pool queueName++-- | Setting a visibility timeout on a message that no longer exists must be an+-- ordinary, reportable outcome rather than a session failure.+--+-- @pgmq.set_vt@ is @RETURNS SETOF@ and yields zero rows for an absent+-- @msg_id@. Decoding that with a single-row decoder produces hasql's+-- @UnexpectedRowCountStatementError@ wrapped in a @StatementSessionError@ —+-- the same shape a genuine infrastructure failure has — so a caller extending+-- a lease could not distinguish "someone else already deleted this message"+-- from "the database is broken".+--+-- Both functions therefore return @Maybe Message@: @Nothing@ for an absent+-- row, @Just@ for a live one, and a session error only for a genuine failure.+testSetVtOnMissingRow :: Pool.Pool -> TestTree+testSetVtOnMissingRow p = testCase "set_vt on a raced-away row returns Nothing" $ do+  withTestFixture p $ \TestFixture {pool, queueName} -> do+    assertSession pool (Sessions.createQueue queueName)+    let missingId = MessageId 999999+    futureTime <- addUTCTime 60 <$> getCurrentTime+    missingChanged <-+      assertSession pool $+        Sessions.changeVisibilityTimeout+          VisibilityTimeoutQuery+            { queueName = queueName,+              messageId = missingId,+              visibilityTimeoutOffset = 60+            }+    assertEqual "changeVisibilityTimeout on a missing message should be Nothing" Nothing (fmap PgmqTypes.messageId missingChanged)+    missingSetAt <-+      assertSession pool $+        Sessions.setVisibilityTimeoutAt+          VisibilityTimeoutAtQuery+            { queueName = queueName,+              messageId = missingId,+              visibilityTime = futureTime+            }+    assertEqual "setVisibilityTimeoutAt on a missing message should be Nothing" Nothing (fmap PgmqTypes.messageId missingSetAt)+    -- An existing message must still be updated and returned.+    msgId <-+      assertSession pool $+        Sessions.sendMessage+          SendMessage+            { queueName = queueName,+              messageBody = MessageBody (object ["vt" .= ("present" :: Text)]),+              delay = Nothing+            }+    changed <-+      assertJust+        =<< assertSession+          pool+          ( Sessions.changeVisibilityTimeout+              VisibilityTimeoutQuery+                { queueName = queueName,+                  messageId = msgId,+                  visibilityTimeoutOffset = 60+                }+          )+    assertEqual "changeVisibilityTimeout should return the message" msgId (PgmqTypes.messageId changed)+    setAt <-+      assertJust+        =<< assertSession+          pool+          ( Sessions.setVisibilityTimeoutAt+              VisibilityTimeoutAtQuery+                { queueName = queueName,+                  messageId = msgId,+                  visibilityTime = futureTime+                }+          )+    assertEqual "setVisibilityTimeoutAt should return the message" msgId (PgmqTypes.messageId setAt)+    cleanupQueue pool queueName

@@ -4,8 +4,13 @@ module QueueSpec (tests) where  import EphemeralDb (TestFixture (..), withTestFixture)+import Hasql.Decoders qualified as D+import Hasql.Encoders qualified as E import Hasql.Pool qualified as Pool+import Hasql.Session qualified as Session+import Hasql.Statement (Statement, preparable) import Pgmq.Hasql.Sessions qualified as Sessions+import Pgmq.Hasql.Statements.Types qualified as StmtTypes import Pgmq.Types (Queue (..), parseQueueName) import Test.Tasty (TestTree, testGroup) import Test.Tasty.HUnit (assertBool, testCase, (@?=))@@ -24,8 +29,11 @@       testDropQueue p,       testDropNonExistentQueue p,       testListQueues p,-      testCreateUnloggedQueue p-      -- Note: testCreatePartitionedQueue is skipped because it requires pg_partman extension+      testCreateUnloggedQueue p,+      -- Note: the partitioned-queue tests need the pg_partman extension, which+      -- is not present in the ephemeral test environment; this one reports the+      -- skip rather than pretending to have run.+      testCreatePartitionedQueueIsReentrant p     ]  testCreateQueue :: Pool.Pool -> TestTree@@ -74,6 +82,34 @@   -- Cleanup   cleanupQueue p queueName1   cleanupQueue p queueName2++-- | Two replicas can call @create_partitioned@ for the same queue: the advisory+-- lock serializes them, but the second one used to fail anyway because+-- @partman.create_parent@ rejects an already-registered parent. Migration+-- 0003 guards both @create_parent@ calls with a @part_config@ probe.+testCreatePartitionedQueueIsReentrant :: Pool.Pool -> TestTree+testCreatePartitionedQueueIsReentrant p =+  testCase "createPartitionedQueue is re-entrant (needs pg_partman)" $ do+    available <- assertSession p (Session.statement () pgPartmanAvailable)+    if not available+      then putStrLn "    SKIPPED: pg_partman is not available in this PostgreSQL installation"+      else do+        qName <- assertRight $ parseQueueName "test_partitioned_reentry"+        let request =+              StmtTypes.CreatePartitionedQueue+                { StmtTypes.queueName = qName,+                  StmtTypes.partitionInterval = "10000",+                  StmtTypes.retentionInterval = "100000"+                }+        assertSession p (Sessions.createPartitionedQueue request)+        assertSession p (Sessions.createPartitionedQueue request)+        cleanupQueue p qName++pgPartmanAvailable :: Statement () Bool+pgPartmanAvailable = preparable sql E.noParams decoder+  where+    sql = "select exists (select 1 from pg_available_extensions where name = 'pg_partman')"+    decoder = D.singleRow (D.column (D.nonNullable D.bool))  testCreateUnloggedQueue :: Pool.Pool -> TestTree testCreateUnloggedQueue p = testCase "createUnloggedQueue creates an unlogged queue" $ do