diff --git a/CHANGELOG.md b/CHANGELOG.md --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -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__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_`; the real channel is + `pgmq.q_.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. diff --git a/pgmq-hasql.cabal b/pgmq-hasql.cabal --- a/pgmq-hasql.cabal +++ b/pgmq-hasql.cabal @@ -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 diff --git a/src/Pgmq.hs b/src/Pgmq.hs --- a/src/Pgmq.hs +++ b/src/Pgmq.hs @@ -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, diff --git a/src/Pgmq/Hasql/Decoders.hs b/src/Pgmq/Hasql/Decoders.hs --- a/src/Pgmq/Hasql/Decoders.hs +++ b/src/Pgmq/Hasql/Decoders.hs @@ -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) diff --git a/src/Pgmq/Hasql/Encoders.hs b/src/Pgmq/Hasql/Encoders.hs --- a/src/Pgmq/Hasql/Encoders.hs +++ b/src/Pgmq/Hasql/Encoders.hs @@ -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 diff --git a/src/Pgmq/Hasql/Sessions.hs b/src/Pgmq/Hasql/Sessions.hs --- a/src/Pgmq/Hasql/Sessions.hs +++ b/src/Pgmq/Hasql/Sessions.hs @@ -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 diff --git a/src/Pgmq/Hasql/Statements/Message.hs b/src/Pgmq/Hasql/Statements/Message.hs --- a/src/Pgmq/Hasql/Statements/Message.hs +++ b/src/Pgmq/Hasql/Statements/Message.hs @@ -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+) diff --git a/src/Pgmq/Hasql/Statements/QueueManagement.hs b/src/Pgmq/Hasql/Statements/QueueManagement.hs --- a/src/Pgmq/Hasql/Statements/QueueManagement.hs +++ b/src/Pgmq/Hasql/Statements/QueueManagement.hs @@ -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_ +-- +-- Notifications are sent via PostgreSQL LISTEN\/NOTIFY on the channel computed by +-- 'Pgmq.Types.notifyChannelName' — @pgmq.q_\.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 () diff --git a/src/Pgmq/Hasql/Statements/QueueObservability.hs b/src/Pgmq/Hasql/Statements/QueueObservability.hs --- a/src/Pgmq/Hasql/Statements/QueueObservability.hs +++ b/src/Pgmq/Hasql/Statements/QueueObservability.hs @@ -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_\_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 diff --git a/src/Pgmq/Hasql/Statements/Types.hs b/src/Pgmq/Hasql/Statements/Types.hs --- a/src/Pgmq/Hasql/Statements/Types.hs +++ b/src/Pgmq/Hasql/Statements/Types.hs @@ -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) diff --git a/test/AdvancedOpsSpec.hs b/test/AdvancedOpsSpec.hs --- a/test/AdvancedOpsSpec.hs +++ b/test/AdvancedOpsSpec.hs @@ -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 diff --git a/test/AliasingSpec.hs b/test/AliasingSpec.hs new file mode 100644 --- /dev/null +++ b/test/AliasingSpec.hs @@ -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_. + 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)) diff --git a/test/AllFunctionsDecoderSpec.hs b/test/AllFunctionsDecoderSpec.hs --- a/test/AllFunctionsDecoderSpec.hs +++ b/test/AllFunctionsDecoderSpec.hs @@ -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) diff --git a/test/EphemeralDb.hs b/test/EphemeralDb.hs --- a/test/EphemeralDb.hs +++ b/test/EphemeralDb.hs @@ -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 diff --git a/test/Main.hs b/test/Main.hs --- a/test/Main.hs +++ b/test/Main.hs @@ -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, diff --git a/test/MessageSpec.hs b/test/MessageSpec.hs --- a/test/MessageSpec.hs +++ b/test/MessageSpec.hs @@ -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 diff --git a/test/MixedCaseRemediationSpec.hs b/test/MixedCaseRemediationSpec.hs new file mode 100644 --- /dev/null +++ b/test/MixedCaseRemediationSpec.hs @@ -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)) diff --git a/test/NotifyChannelSpec.hs b/test/NotifyChannelSpec.hs new file mode 100644 --- /dev/null +++ b/test/NotifyChannelSpec.hs @@ -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_\@, so anyone +-- following them listened on a channel that never receives anything. The real +-- channel is @pgmq.q_\.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) diff --git a/test/NotifyRaceSpec.hs b/test/NotifyRaceSpec.hs new file mode 100644 --- /dev/null +++ b/test/NotifyRaceSpec.hs @@ -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 diff --git a/test/NullBodySpec.hs b/test/NullBodySpec.hs new file mode 100644 --- /dev/null +++ b/test/NullBodySpec.hs @@ -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)))) diff --git a/test/NullSemanticsSpec.hs b/test/NullSemanticsSpec.hs new file mode 100644 --- /dev/null +++ b/test/NullSemanticsSpec.hs @@ -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 diff --git a/test/QueueSpec.hs b/test/QueueSpec.hs --- a/test/QueueSpec.hs +++ b/test/QueueSpec.hs @@ -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