pgmq-hasql 0.6.1.1 → 0.7.0.0

raw patch · 23 files changed

+1260/−31 lines, 23 filesdep +bytestringdep +containersdep +tasty-goldendep ~ephemeral-pgdep ~pgmq-coredep ~pgmq-hasqlPVP ok

version bump matches the API change (PVP)

Dependencies added: bytestring, containers, tasty-golden

Dependency ranges changed: ephemeral-pg, pgmq-core, pgmq-hasql, pgmq-migration

API changes (from Hackage documentation)

Files

• CHANGELOG.md +34/−0
• pgmq-hasql.cabal +19/−7
• src/Pgmq.hs +25/−1
• src/Pgmq/Hasql/Decoders.hs +13/−1
• src/Pgmq/Hasql/Sessions.hs +58/−1
• src/Pgmq/Hasql/Statements.hs +3/−1
• src/Pgmq/Hasql/Statements/Inspection.hs +110/−0
• src/Pgmq/Hasql/Statements/QueueManagement.hs +4/−0
• src/Pgmq/Hasql/Statements/QueueObservability.hs +18/−9
• src/Pgmq/Hasql/Statements/Types.hs +68/−2
• test/AliasingSpec.hs +3/−1
• test/EphemeralDb.hs +28/−3
• test/InspectionCostSpec.hs +120/−0
• test/InspectionForeignNameSpec.hs +119/−0
• test/InspectionSpec.hs +364/−0
• test/JsonGoldenSpec.hs +76/−0
• test/Main.hs +17/−2
• test/NotifyChannelSpec.hs +4/−2
• test/PartitionedNotifySpec.hs +101/−0
• test/RootPermissionsSpec.hs +63/−0
• test/UmbrellaExportsSpec.hs +11/−1
• test/golden/queue_metrics.json +1/−0
• test/golden/queue_metrics_partitioned.json +1/−0

@@ -1,5 +1,39 @@ # Revision history for pgmq-hasql +## 0.7.0.0 -- 2026-10-26++### New Features++- Add non-destructive inspection reads, exported from `Pgmq` and+  `Pgmq.Hasql.Sessions`: `peekMessages`, `peekArchivedMessages`, `lookupMessage`,+  `lookupArchivedMessage` and `queueMetricsUnvalidated`. Paging uses ascending+  `msg_id` with an exclusive cursor and a row limit, never `OFFSET`; lookups return+  `Nothing` for missing messages. Reads leave `vt`, `read_ct` and `last_read_at`+  untouched.+- `PeekMessages` and `LookupMessage` accept plain `Text` names, including foreign and+  mixed-case names that `parseQueueName` rejects. Physical tables are resolved with+  `pgmq.format_table_name` and quoted as identifiers; missing queues retain server+  error `42P01`.+- Add `Pgmq.Hasql.Statements.Inspection` with `formatTableName`, `quoteIdentifier`+  and four `unpreparable` statement builders, plus `archivedMessageDecoder` in+  `Pgmq.Hasql.Decoders`.+- Add `ToJSON QueueMetrics` with snake_case fields and explicit nulls. Total depth+  (`queue_length`) stays distinct from available depth (`queue_visible_length`);+  null `default_partition_length` means unavailable/inapplicable, not zero.+  Goldens and parity properties pin the published contract in design note 020.++### Other Changes++- Document partition notification silence and cover native/stock partition channels,+  native aliases and ordinary throttling.+- Inspection tests cover late lower-ID commits and late archival, propagate consumer+  failures and exercise 100,000-row ordinary/unlogged and numeric/time-partitioned+  live/archive layouts. Design note 019 records query plans, moving-view semantics,+  finite database deadlines and separate inspection capacity requirements. Row+  limits do not bound scan work or payload bytes.+- Ephemeral PostgreSQL roots are created/validated as owner-only (`0700`), existing+  owned roots are tightened, unsafe paths are rejected and registry contents survive.+ ## 0.6.1.1 -- 2026-09-25  Coordinated family version bump; the library is unchanged from 0.6.1.0.

@@ -1,6 +1,6 @@ cabal-version: 3.4 name: pgmq-hasql-version: 0.6.1.1+version: 0.7.0.0 synopsis: Hasql-based client for PGMQ (PostgreSQL Message Queue) description:   A Haskell client library for PGMQ (PostgreSQL Message Queue) built@@ -17,7 +17,9 @@ category: Database build-type: Simple extra-doc-files: CHANGELOG.md-extra-source-files: test/fixtures/pgmq-1.12.0.sql+extra-source-files:+  test/fixtures/pgmq-1.12.0.sql+  test/golden/*.json  common warnings   ghc-options:@@ -42,6 +44,7 @@     Pgmq.Hasql.Quasi     Pgmq.Hasql.Sessions     Pgmq.Hasql.Statements+    Pgmq.Hasql.Statements.Inspection     Pgmq.Hasql.Statements.Message     Pgmq.Hasql.Statements.QueueManagement     Pgmq.Hasql.Statements.QueueObservability@@ -64,7 +67,7 @@     hasql ^>=1.10,     hasql-transaction ^>=1.2,     lens ^>=5.3,-    pgmq-core >=0.6 && <0.7,+    pgmq-core >=0.7 && <0.8,     template-haskell >=2.20 && <3,     text ^>=2.1,     time ^>=1.14,@@ -91,6 +94,10 @@     DecoderValidationSpec     EphemeralDb     Generators+    InspectionCostSpec+    InspectionForeignNameSpec+    InspectionSpec+    JsonGoldenSpec     MessageSpec     MetricsSpec     MixedCaseRemediationSpec@@ -98,7 +105,9 @@     NotifyRaceSpec     NullBodySpec     NullSemanticsSpec+    PartitionedNotifySpec     QueueSpec+    RootPermissionsSpec     RoundTripSpec     SchemaSpec     TestUtils@@ -112,19 +121,22 @@   build-depends:     aeson,     base >=4.18 && <5,+    bytestring >=0.11 && <0.13,+    containers,     directory,-    ephemeral-pg >=0.3.1.0,+    ephemeral-pg >=0.3.1 && <0.4,     hasql,     hasql-pool ^>=1.4,     hedgehog ^>=1.5,     pg-migrate,-    pgmq-core >=0.6 && <0.7,-    pgmq-hasql >=0.6 && <0.7,-    pgmq-migration >=0.6 && <0.7,+    pgmq-core >=0.7 && <0.8,+    pgmq-hasql >=0.7 && <0.8,+    pgmq-migration >=0.7 && <0.8,     postgresql-libpq >=0.10.1 && <0.12,     random ^>=1.2,     scientific ^>=0.3,     tasty ^>=1.5,+    tasty-golden ^>=2.3.6,     tasty-hedgehog ^>=1.4,     tasty-hunit ^>=0.10,     text,

@@ -56,6 +56,21 @@     ReadGrouped (..),     ReadGroupedWithPoll (..), +    -- * Non-destructive Inspection++    -- | Reads that observe a queue or its archive without leasing anything:+    -- @vt@ and @read_ct@ are untouched. They accept any server-accepted name+    -- (pass 'queueNameToText' for a validated one) and page by exclusive+    -- @msg_id@ cursor, never @OFFSET@. See+    -- @docs/design/019-non-destructive-inspection-reads.md@.+    peekMessages,+    peekArchivedMessages,+    lookupMessage,+    lookupArchivedMessage,+    queueMetricsUnvalidated,+    PeekMessages (..),+    LookupMessage (..),+     -- * Topic Routing (pgmq 1.11.0+)      -- ** Topic Management@@ -84,6 +99,7 @@     MessageHeaders (..),     MessageId (..),     Message (..),+    ArchivedMessage (..),     Queue (..),     UnvalidatedQueue (..),     QueueName,@@ -173,8 +189,13 @@     listQueuesUnvalidated,     listTopicBindings,     listTopicBindingsForQueue,+    lookupArchivedMessage,+    lookupMessage,+    peekArchivedMessages,+    peekMessages,     pop,     queueMetrics,+    queueMetricsUnvalidated,     readGrouped,     readGroupedHead,     readGroupedHeadWithPoll,@@ -211,7 +232,9 @@     BindTopic (..),     CreatePartitionedQueue (..),     EnableNotifyInsert (..),+    LookupMessage (..),     MessageQuery (..),+    PeekMessages (..),     PopMessage (..),     QueueMetrics (..),     ReadGrouped (..),@@ -230,7 +253,8 @@     VisibilityTimeoutQuery (..),   ) import Pgmq.Types-  ( Message (..),+  ( ArchivedMessage (..),+    Message (..),     MessageBody (..),     MessageHeaders (..),     MessageId (..),

@@ -1,6 +1,7 @@ module Pgmq.Hasql.Decoders   ( messageDecoder,     messageIdDecoder,+    archivedMessageDecoder,     queueDecoder,     unvalidatedQueueDecoder,     queueMetricsDecoder,@@ -19,7 +20,8 @@ import Hasql.Decoders qualified as D import Pgmq.Hasql.Statements.Types (QueueMetrics (..)) import Pgmq.Types-  ( Message (..),+  ( ArchivedMessage (..),+    Message (..),     MessageBody (..),     MessageId (..),     NotifyInsertThrottle (..),@@ -65,6 +67,16 @@  messageIdDecoder :: D.Row MessageId messageIdDecoder = MessageId <$> D.column (D.nonNullable D.int8)++-- | Decoder for an archive-table row projected as the seven message columns in+-- 'messageDecoder' order followed by @archived_at@. The inspection statements+-- list the columns explicitly in exactly this order; the archive table's own+-- column order (which puts @archived_at@ before @vt@) is irrelevant.+archivedMessageDecoder :: D.Row ArchivedMessage+archivedMessageDecoder =+  ArchivedMessage+    <$> messageDecoder+    <*> D.column (D.nonNullable D.timestamptz) -- archived_at  -- | Decoder for pgmq.queue_record type -- Column order: queue_name (varchar), is_partitioned (bool), is_unlogged (bool), created_at (timestamptz)

@@ -33,6 +33,12 @@     allQueueMetrics,     readMessage,     readWithPoll,+    -- Non-destructive inspection (no upstream function)+    peekMessages,+    peekArchivedMessages,+    lookupMessage,+    lookupArchivedMessage,+    queueMetricsUnvalidated,     -- FIFO read functions (pgmq 1.8.0+)     readGrouped,     readGroupedWithPoll,@@ -69,6 +75,7 @@ import Hasql.Session (Session, statement) import Pgmq.Hasql.Prelude import Pgmq.Hasql.Statements qualified as Stmt+import Pgmq.Hasql.Statements.Inspection qualified as Inspect import Pgmq.Hasql.Statements.Message qualified as Msg import Pgmq.Hasql.Statements.TopicManagement qualified as Topic import Pgmq.Hasql.Statements.Types@@ -86,7 +93,9 @@     BindTopic,     CreatePartitionedQueue,     EnableNotifyInsert,+    LookupMessage (..),     MessageQuery,+    PeekMessages (..),     PopMessage,     QueueMetrics,     ReadGrouped,@@ -105,7 +114,8 @@     VisibilityTimeoutQuery,   ) import Pgmq.Types-  ( Message,+  ( ArchivedMessage,+    Message,     MessageId,     NotifyInsertThrottle,     Queue,@@ -216,6 +226,10 @@ detachArchive q = statement q Stmt.detachArchive  -- | Enable insert notifications for a queue (pgmq 1.7.0+)+--+-- Normal partitioned queues receive no insert notifications on their documented+-- parent channel. See @docs/design/015-notification-delivery-contract.md@.+-- The reconciler reports partitioned notification declarations as unsupported. enableNotifyInsert :: EnableNotifyInsert -> Session () enableNotifyInsert config = statement config Stmt.enableNotifyInsert @@ -232,6 +246,49 @@  allQueueMetrics :: Session [QueueMetrics] allQueueMetrics = statement () Stmt.allQueueMetrics++-- | Metrics for a queue named by plain text, including names+-- 'Pgmq.Types.parseQueueName' rejects. Same projection as 'queueMetrics'.+queueMetricsUnvalidated :: Text -> Session QueueMetrics+queueMetricsUnvalidated q = statement q Stmt.queueMetricsUnvalidated++-- Non-destructive inspection ---------------------------------------------------+--+-- Each read resolves the physical table on the server first, then runs an+-- unprepared SELECT against it. Neither statement modifies a row: vt, read_ct,+-- and last_read_at are exactly what they were. A missing queue surfaces as the+-- server's undefined_table error (SQLSTATE 42P01); a missing message is Nothing.++-- | A keyset page of a queue table in ascending @msg_id@ order, starting+-- strictly after 'afterMessageId', without leasing anything. Each page has its+-- own snapshot; refresh from the beginning to find lower IDs committed later.+-- The row limit does not bound scan work: partitioned queues may scan and sort.+-- Use finite database deadlines and capacity separate from worker connections.+peekMessages :: PeekMessages -> Session (Vector Message)+peekMessages PeekMessages {unvalidatedQueueName, afterMessageId, limit} = do+  table <- statement (unvalidatedQueueName, "q") Inspect.formatTableName+  statement (afterMessageId, limit) (Inspect.peekStatement table)++-- | A keyset page of an archive table, each row with its archival timestamp.+-- An older ID archived after the cursor advances requires a first-page refresh.+-- The same cost and isolation limits as 'peekMessages' apply.+peekArchivedMessages :: PeekMessages -> Session (Vector ArchivedMessage)+peekArchivedMessages PeekMessages {unvalidatedQueueName, afterMessageId, limit} = do+  table <- statement (unvalidatedQueueName, "a") Inspect.formatTableName+  statement (afterMessageId, limit) (Inspect.peekArchivedStatement table)++-- | One message of a queue table by id, or 'Nothing' when it is not there+-- (deleted, archived, popped, or never sent).+lookupMessage :: LookupMessage -> Session (Maybe Message)+lookupMessage LookupMessage {unvalidatedQueueName, messageId} = do+  table <- statement (unvalidatedQueueName, "q") Inspect.formatTableName+  statement messageId (Inspect.lookupStatement table)++-- | One message of an archive table by id, or 'Nothing'.+lookupArchivedMessage :: LookupMessage -> Session (Maybe ArchivedMessage)+lookupArchivedMessage LookupMessage {unvalidatedQueueName, messageId} = do+  table <- statement (unvalidatedQueueName, "a") Inspect.formatTableName+  statement messageId (Inspect.lookupArchivedStatement table)  readMessage :: ReadMessage -> Session (Vector Message) readMessage query = statement query Stmt.readMessage

@@ -1,11 +1,13 @@ module Pgmq.Hasql.Statements-  ( module Pgmq.Hasql.Statements.Message,+  ( module Pgmq.Hasql.Statements.Inspection,+    module Pgmq.Hasql.Statements.Message,     module Pgmq.Hasql.Statements.QueueManagement,     module Pgmq.Hasql.Statements.QueueObservability,     module Pgmq.Hasql.Statements.TopicManagement,   ) where +import Pgmq.Hasql.Statements.Inspection import Pgmq.Hasql.Statements.Message import Pgmq.Hasql.Statements.QueueManagement import Pgmq.Hasql.Statements.QueueObservability

@@ -0,0 +1,110 @@+-- | Non-destructive inspection statements over queue and archive tables.+--+-- Upstream pgmq offers no read that does not lease (@pgmq.read@,+-- @read_with_poll@, and @pop@ all set @vt@ and bump @read_ct@), no read of an+-- archive table, and no fetch by id. These statements are plain @SELECT@s over+-- the tables the upstream schema defines; they shadow no @pgmq.*@ function and+-- add no migration, which is exactly the kind of hand-written statement+-- @docs/design/012-vendor-upstream-pgmq-sql.md@ permits. None of them modifies+-- a row.+--+-- The physical table is resolved on the server by 'formatTableName' (upstream's+-- @pgmq.format_table_name@, which lowercases and rejects @$@, @;@, @--@, and+-- @'@) and spliced into the SQL as a double-quoted identifier by+-- 'quoteIdentifier'. Because the SQL text differs per table, every builder uses+-- 'unpreparable': hasql caches prepared statements by SQL text, and a prepared+-- statement per queue would grow the per-connection cache without bound.+--+-- Pages are keyset pages: @msg_id > cursor@, ordered by @msg_id@, bounded by+-- @LIMIT@, never @OFFSET@. A limit bounds returned rows, not scan work;+-- partitioned layouts may scan and sort. Separate calls do not hold a shared+-- snapshot: refresh from the first page for lower IDs committed or archived+-- after a cursor advances. Use finite statement/lock deadlines and separate+-- inspection capacity. See @docs/design/019-non-destructive-inspection-reads.md@.+module Pgmq.Hasql.Statements.Inspection+  ( formatTableName,+    quoteIdentifier,+    peekStatement,+    peekArchivedStatement,+    lookupStatement,+    lookupArchivedStatement,+  )+where++import Data.Text qualified as T+import Hasql.Decoders qualified as D+import Hasql.Encoders qualified as E+import Hasql.Statement (Statement, preparable, unpreparable)+import Pgmq.Hasql.Decoders (archivedMessageDecoder, messageDecoder)+import Pgmq.Hasql.Encoders (messageIdValue)+import Pgmq.Hasql.Prelude+import Pgmq.Types (ArchivedMessage, Message, MessageId)++-- | @select pgmq.format_table_name($1, $2)@: the physical table name for a+-- queue name and a prefix (@"q"@ for the queue table, @"a"@ for the archive).+-- Raises the server's own error for names containing @$@, @;@, @--@, or @'@.+formatTableName :: Statement (Text, Text) Text+formatTableName = preparable sql encoder decoder+  where+    sql = "select pgmq.format_table_name($1, $2)"+    encoder =+      (fst >$< E.param (E.nonNullable E.text))+        <> (snd >$< E.param (E.nonNullable E.text))+    decoder = D.singleRow (D.column (D.nonNullable D.text))++-- | Double-quote a SQL identifier, doubling any embedded double quote, so a+-- physical table name such as @q_odd-name@ or @q_myqueue@ can be spliced into+-- SQL text safely.+quoteIdentifier :: Text -> Text+quoteIdentifier ident = "\"" <> T.replace "\"" "\"\"" ident <> "\""++-- | The seven columns 'messageDecoder' expects, in its order.+messageColumns :: Text+messageColumns = "msg_id, read_ct, enqueued_at, last_read_at, vt, message, headers"++-- | 'messageColumns' followed by the archive table's @archived_at@.+archivedColumns :: Text+archivedColumns = messageColumns <> ", archived_at"++pageEncoder :: E.Params (Maybe MessageId, Int32)+pageEncoder =+  (fst >$< E.param (E.nullable messageIdValue))+    <> (snd >$< E.param (E.nonNullable E.int4))++messageIdEncoder :: E.Params MessageId+messageIdEncoder = E.param (E.nonNullable messageIdValue)++-- | A keyset page of a queue table. The argument is the already-resolved+-- physical table name (from 'formatTableName' with prefix @"q"@). Returns at+-- most @limit@ rows with @msg_id@ strictly greater than the cursor, in+-- ascending @msg_id@ order, and modifies nothing.+peekStatement :: Text -> Statement (Maybe MessageId, Int32) (Vector Message)+peekStatement table = unpreparable (pageSql messageColumns table) pageEncoder (D.rowVector messageDecoder)++-- | A keyset page of an archive table (from 'formatTableName' with prefix+-- @"a"@), each row carrying @archived_at@. Same paging rules as+-- 'peekStatement'.+peekArchivedStatement :: Text -> Statement (Maybe MessageId, Int32) (Vector ArchivedMessage)+peekArchivedStatement table = unpreparable (pageSql archivedColumns table) pageEncoder (D.rowVector archivedMessageDecoder)++-- | One row of a queue table by id, or 'Nothing'. Modifies nothing.+lookupStatement :: Text -> Statement MessageId (Maybe Message)+lookupStatement table = unpreparable (lookupSql messageColumns table) messageIdEncoder (D.rowMaybe messageDecoder)++-- | One row of an archive table by id, or 'Nothing'.+lookupArchivedStatement :: Text -> Statement MessageId (Maybe ArchivedMessage)+lookupArchivedStatement table = unpreparable (lookupSql archivedColumns table) messageIdEncoder (D.rowMaybe archivedMessageDecoder)++-- | @$1@ is the exclusive cursor (nullable), @$2@ the page size. The explicit+-- null test keeps the no-cursor case independent of where ids start.+pageSql :: Text -> Text -> Text+pageSql columns table =+  "select "+    <> columns+    <> " from pgmq."+    <> quoteIdentifier table+    <> " where ($1::bigint is null or msg_id > $1) order by msg_id asc limit $2"++lookupSql :: Text -> Text -> Text+lookupSql columns table =+  "select " <> columns <> " from pgmq." <> quoteIdentifier table <> " where msg_id = $1"

@@ -74,6 +74,10 @@ -- 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.+--+-- Normal partitioned queues receive no insert notifications on their documented+-- parent channel. See @docs/design/015-notification-delivery-contract.md@.+-- The reconciler reports partitioned notification declarations as unsupported. enableNotifyInsert :: Statement EnableNotifyInsert () enableNotifyInsert = preparable sql enableNotifyInsertEncoder D.noResult   where

@@ -3,6 +3,7 @@     listQueuesUnvalidated,     listFifoIndexQueueNames,     queueMetrics,+    queueMetricsUnvalidated,     allQueueMetrics,   ) where@@ -50,18 +51,26 @@       \where schemaname = 'pgmq' and indexname ~ '^q_.*_fifo_idx$'"     decoder = D.rowList (D.column (D.nonNullable D.text)) +-- | The metrics projection with a stable shape on pgmq 1.12 and 1.13. The JSON+-- record lookup preserves SQL NULL for the attribute 1.12 lacks.+queueMetricsSql :: Text+queueMetricsSql =+  "select m.queue_name, m.queue_length, m.newest_msg_age_sec, m.oldest_msg_age_sec, \+  \m.total_messages, m.scrape_time, m.queue_visible_length, \+  \(to_jsonb(m)->>'default_partition_length')::bigint \+  \from pgmq.metrics($1) as m"+ -- | Metrics with a stable projection on pgmq 1.12 and 1.13.--- The JSON record lookup preserves SQL NULL for the missing 1.12 attribute. -- | https://pgmq.github.io/pgmq/api/sql/functions/#metrics queueMetrics :: Statement QueueName QueueMetrics-queueMetrics = preparable sql queueNameEncoder decoder-  where-    sql =-      "select m.queue_name, m.queue_length, m.newest_msg_age_sec, m.oldest_msg_age_sec, \-      \m.total_messages, m.scrape_time, m.queue_visible_length, \-      \(to_jsonb(m)->>'default_partition_length')::bigint \-      \from pgmq.metrics($1) as m"-    decoder = D.singleRow queueMetricsDecoder+queueMetrics = preparable queueMetricsSql queueNameEncoder (D.singleRow queueMetricsDecoder)++-- | 'queueMetrics' for a queue named by plain text: any name the server+-- accepts, including names 'Pgmq.Types.parseQueueName' rejects, so an+-- inspection surface can report metrics for foreign queues. Same SQL, same+-- decoder, same nullable eighth column.+queueMetricsUnvalidated :: Statement Text QueueMetrics+queueMetricsUnvalidated = preparable queueMetricsSql (E.param (E.nonNullable E.text)) (D.singleRow queueMetricsDecoder)  -- | https://pgmq.github.io/pgmq/api/sql/functions/#metrics_all allQueueMetrics :: Statement () [QueueMetrics]

@@ -23,6 +23,9 @@     -- FIFO read types (pgmq 1.8.0+)     ReadGrouped (..),     ReadGroupedWithPoll (..),+    -- Non-destructive inspection (no upstream function)+    PeekMessages (..),+    LookupMessage (..),     -- Topic types (pgmq 1.11.0+)     BindTopic (..),     UnbindTopic (..),@@ -36,8 +39,8 @@   ) where -import Data.Aeson (Value)-import Pgmq.Hasql.Prelude+import Data.Aeson (KeyValue, Value, object, pairs, (.=))+import Pgmq.Hasql.Prelude hiding ((.=)) import Pgmq.Types (MessageBody, MessageHeaders, MessageId, QueueName, RoutingKey, TopicPattern)  type Delay = Int32@@ -318,3 +321,66 @@     throttleIntervalMs :: !Int32   }   deriving stock (Generic)++-- | A keyset page over a queue table (@pgmq.q_\<name\>@) or, through+-- @peekArchivedMessages@, an archive table (@pgmq.a_\<name\>@).+--+-- @unvalidatedQueueName@ is any name the server accepts, including names+-- 'Pgmq.Types.parseQueueName' rejects: an inspection surface must show what+-- exists. Pass 'Pgmq.Types.queueNameToText' when you hold a validated name.+-- The physical table is resolved by @pgmq.format_table_name@ on the server.+--+-- @afterMessageId@ is an /exclusive/ cursor: the page starts strictly after+-- it; 'Nothing' starts at the beginning. @limit@ is passed straight to SQL+-- @LIMIT@, so it must be positive (@0@ returns nothing; a negative value is a+-- server error). Pages are ordered by @msg_id@ ascending and never use+-- @OFFSET@, so deleting earlier rows does not shift the page. Each call has+-- its own snapshot: lower IDs committed or archived later require a refresh+-- from the first page. This is not a replay feed. @LIMIT@ bounds rows returned,+-- not scan work or payload bytes; partitioned layouts may scan and sort. Use+-- finite statement/lock deadlines and inspection capacity separate from workers.+data PeekMessages = PeekMessages+  { unvalidatedQueueName :: !Text,+    afterMessageId :: !(Maybe MessageId),+    limit :: !Int32+  }+  deriving stock (Generic)++-- | One message by id in a queue table or, through @lookupArchivedMessage@, an+-- archive table. The name rules are those of t'PeekMessages'.+data LookupMessage = LookupMessage+  { unvalidatedQueueName :: !Text,+    messageId :: !MessageId+  }+  deriving stock (Generic)++-- JSON wire encodings ---------------------------------------------------------++-- | Wire keys are the @pgmq.metrics_result@ columns. Two of them are easy to+-- confuse and must not be: @queue_length@ is the total number of rows in the+-- queue table, and @queue_visible_length@ is the subset whose visibility+-- timeout has expired and are therefore available to a reader (see+-- @docs/design/002-queue-visible-length.md@). @default_partition_length@ is+-- @null@ when the metric is unavailable (PGMQ 1.12) or inapplicable (a+-- non-partitioned queue); @null@ never means zero. Encode-only. The encoding is+-- a published compatibility surface; see+-- @docs/design/020-json-wire-encodings.md@.+--+-- The pattern is positional because @queueName@ is a duplicate record field in+-- this module and its selector is ambiguous under GHC 9.12.+queueMetricsFields :: (KeyValue e kv) => QueueMetrics -> [kv]+queueMetricsFields (QueueMetrics qn len newest oldest total scrape visible dpl) =+  [ "queue_name" .= qn,+    "queue_length" .= len,+    "newest_msg_age_sec" .= newest,+    "oldest_msg_age_sec" .= oldest,+    "total_messages" .= total,+    "scrape_time" .= scrape,+    "queue_visible_length" .= visible,+    "default_partition_length" .= dpl+  ]++-- | JSON wire contract: @docs/design/020-json-wire-encodings.md@.+instance ToJSON QueueMetrics where+  toJSON = object . queueMetricsFields+  toEncoding = pairs . mconcat . queueMetricsFields

@@ -40,6 +40,7 @@     migrationPlan,     runMigrationPlan,   )+import EphemeralDb (ephemeralConfig) import EphemeralPg qualified as Pg import Hasql.Decoders qualified as D import Hasql.Pool qualified as Pool@@ -169,7 +170,8 @@ -- installed, exactly as @EphemeralDb@ does for the shared one. acquireDb :: IO (Pg.Database, Pool.Pool) acquireDb = do-  started <- Pg.startCached Pg.defaultConfig Pg.defaultCacheConfig+  config <- ephemeralConfig+  started <- Pg.startCached config 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 :| []))

@@ -6,6 +6,7 @@     withPgmqDb,     withPgmqPool,     ephemeralConfig,+    ensureOwnerOnlyRoot,      -- * Test fixtures     TestFixture (..),@@ -17,7 +18,8 @@   ) where -import Control.Monad (filterM, when)+import Control.Monad (filterM, unless, when)+import Data.Bits ((.&.)) import Data.List.NonEmpty (NonEmpty (..)) import Data.Monoid (Last (..)) import Data.Text qualified as T@@ -42,8 +44,11 @@ import Hasql.Session qualified as Session import Pgmq.Migration qualified as Migration import Pgmq.Types (QueueName, parseQueueName)-import System.Directory (createDirectoryIfMissing, doesFileExist)+import System.Directory (doesFileExist) import System.Environment (lookupEnv)+import System.IO.Error (catchIOError, isAlreadyExistsError)+import System.Posix.Directory qualified as Posix+import System.Posix.Files (deviceID, fileID, fileMode, fileOwner, getSymbolicLinkStatus, isDirectory, setFileMode) import System.Posix.User (getEffectiveUserID) import System.Random (randomRIO) @@ -62,11 +67,31 @@   uid <- getEffectiveUserID   pure ("/tmp/ephpg-pgmq-hs-" <> show uid) +-- | Establish a real effective-uid-owned root without deleting its contents.+ensureOwnerOnlyRoot :: FilePath -> IO ()+ensureOwnerOnlyRoot root = do+  catchIOError (Posix.createDirectory root 0o700) $ \err ->+    unless (isAlreadyExistsError err) (ioError err)+  uid <- getEffectiveUserID+  before <- getSymbolicLinkStatus root+  unless (isDirectory before && fileOwner before == uid) $+    ioError (userError ("Unsafe ephemeral PostgreSQL root: " <> root))+  setFileMode root 0o700+  after <- getSymbolicLinkStatus root+  unless+    ( isDirectory after+        && fileOwner after == uid+        && deviceID before == deviceID after+        && fileID before == fileID after+        && fileMode after .&. 0o7777 == 0o700+    )+    (ioError (userError ("Ephemeral PostgreSQL root changed or is not owner-only: " <> root)))+ -- | Cached-startup configuration pinned to 'ephemeralRoot'. ephemeralConfig :: IO Config ephemeralConfig = do   root <- ephemeralRoot-  createDirectoryIfMissing True root+  ensureOwnerOnlyRoot root   pure defaultConfig {temporaryRoot = Last (Just root)}  -- | Run an action with a temporary PostgreSQL database that has pgmq schema installed.

@@ -0,0 +1,120 @@+{-# LANGUAGE OverloadedStrings #-}++-- | Complete projections on real upstream layouts, without installing indexes.+-- Set PGMQ_INSPECTION_PLANS to an absolute directory to retain JSON EXPLAINs.+module InspectionCostSpec (tests) where++import Control.Exception (finally)+import Control.Monad (forM_, when)+import Data.Aeson (Value, object, (.=))+import Data.Aeson qualified as Aeson+import Data.Int (Int64)+import Data.Text (Text)+import Data.Text qualified as T+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.Session qualified as Session+import Hasql.Statement (toSql, unpreparable)+import Pgmq.Hasql.Sessions qualified as Sessions+import Pgmq.Hasql.Statements.Inspection qualified as Inspect+import Pgmq.Hasql.Statements.Types (CreatePartitionedQueue (..), LookupMessage (LookupMessage), PeekMessages (PeekMessages))+import Pgmq.Types (ArchivedMessage (..), Message (..), MessageId (..), queueNameToText)+import System.Directory (createDirectoryIfMissing)+import System.Environment (lookupEnv)+import Test.Tasty (TestTree, testGroup)+import Test.Tasty.HUnit (assertBool, assertFailure, testCase, (@?=))+import TestUtils (assertSession, cleanupQueue)++tests :: Pool.Pool -> TestTree+tests pool =+  testGroup "Inspection cost and partition layouts" $+    map (layoutTest pool) ["ordinary", "unlogged", "numeric", "time"]++layoutTest :: Pool.Pool -> Text -> TestTree+layoutTest p layout = testCase (T.unpack layout <> ": 100000 live and archive rows, complete projections") $ do+  available <- assertSession p (scalarBool "select exists (select from pg_extension where extname = 'pg_partman')")+  required <- (== Just "1") <$> lookupEnv "PGMQ_REQUIRE_PARTMAN"+  let partitioned = layout == "numeric" || layout == "time"+  if partitioned && not available+    then if required then assertFailure "required pg_partman is absent" else putStrLn "    SKIPPED: pg_partman is not installed"+    else withTestFixture p $ \TestFixture {pool, queueName} -> flip finally (cleanupQueue pool queueName) $ do+      case layout of+        "ordinary" -> assertSession pool (Sessions.createQueue queueName)+        "unlogged" -> assertSession pool (Sessions.createUnloggedQueue queueName)+        "numeric" -> assertSession pool (Sessions.createPartitionedQueue (CreatePartitionedQueue queueName "10000" "1000000"))+        "time" -> assertSession pool (Sessions.createPartitionedQueue (CreatePartitionedQueue queueName "1 day" "1000 days"))+        _ -> assertFailure "unknown test layout"+      let q = queueNameToText queueName+      forM_ ["q", "a"] $ \prefix -> do+        table <- assertSession pool (statement (q, prefix) Inspect.formatTableName)+        let qualified = "pgmq." <> Inspect.quoteIdentifier table+            timestamp = if layout == "time" then "date_trunc('day', now()) + case n % 3 when 0 then interval '-1 day' when 1 then interval '0 days' else interval '100 years' end" else "now()"+            columns = "msg_id, vt, enqueued_at, message, headers" <> if prefix == "a" then ", archived_at" else ""+            values = "n, now(), " <> timestamp <> ", jsonb_build_object('n', n, 'body', repeat('x', 128)), '{\"fixture\":true}'::jsonb" <> if prefix == "a" then ", " <> timestamp else ""+        assertSession pool $+          Session.script $+            "insert into " <> qualified <> " (" <> columns <> ") overriding system value select " <> values <> " from generate_series(1, 100000) n; analyze " <> qualified+        when partitioned $ do+          (leaves, hasDefault) <-+            assertSession pool $+              statement () $+                unpreparable+                  ("select count(distinct t.tableoid), bool_or(pg_get_expr(c.relpartbound, c.oid) = 'DEFAULT') from " <> qualified <> " t join pg_class c on c.oid = t.tableoid")+                  mempty+                  (D.singleRow ((,) <$> D.column (D.nonNullable D.int8) <*> D.column (D.nonNullable D.bool)))+          assertBool "rows occupy multiple leaves" (leaves >= (3 :: Int64))+          assertBool "rows occupy the default partition" hasDefault+        verifyReads pool q prefix+        let pageSql = if prefix == "q" then toSql (Inspect.peekStatement table) else toSql (Inspect.peekArchivedStatement table)+            lookupSql = if prefix == "q" then toSql (Inspect.lookupStatement table) else toSql (Inspect.lookupArchivedStatement table)+            queries :: [(Text, Text)]+            queries =+              [ ("first", T.replace "$2" "50" (T.replace "$1" "NULL" pageSql)),+                ("late", T.replace "$2" "50" (T.replace "$1" "99950" pageSql)),+                ("present", T.replace "$1" "99999" lookupSql),+                ("absent", T.replace "$1" "100001" lookupSql)+              ]+        plans <-+          mapM+            ( \(label, sql) -> do+                plan <- assertSession pool (explain sql)+                pure (object ["case" .= label, "sql" .= sql, "explain" .= plan])+            )+            queries+        output <- lookupEnv "PGMQ_INSPECTION_PLANS"+        case output of+          Nothing -> pure ()+          Just directory -> do+            createDirectoryIfMissing True directory+            version <- maybe "1.13.0" id <$> lookupEnv "PGMQ_TEST_SCHEMA_VERSION"+            postgres <- assertSession pool (scalarText "select version()")+            partman <- assertSession pool (scalarText "select coalesce((select extversion from pg_extension where extname = 'pg_partman'), 'absent')")+            Aeson.encodeFile (directory <> "/" <> version <> "-" <> T.unpack layout <> "-" <> T.unpack prefix <> ".json") $+              object ["schema" .= version, "postgres" .= postgres, "pg_partman" .= partman, "layout" .= layout, "table" .= table, "rows" .= (100000 :: Int), "plans" .= plans]++verifyReads :: Pool.Pool -> Text -> Text -> IO ()+verifyReads pool queue prefix = do+  let page cursor =+        if prefix == "q"+          then map messageId . V.toList <$> assertSession pool (Sessions.peekMessages (PeekMessages queue cursor 50))+          else map (messageId . archivedMessage) . V.toList <$> assertSession pool (Sessions.peekArchivedMessages (PeekMessages queue cursor 50))+      lookupId mid =+        if prefix == "q"+          then fmap messageId <$> assertSession pool (Sessions.lookupMessage (LookupMessage queue mid))+          else fmap (messageId . archivedMessage) <$> assertSession pool (Sessions.lookupArchivedMessage (LookupMessage queue mid))+  page Nothing >>= (@?= map MessageId [1 .. 50])+  page (Just (MessageId 99950)) >>= (@?= map MessageId [99951 .. 100000])+  lookupId (MessageId 99999) >>= (@?= Just (MessageId 99999))+  lookupId (MessageId 100001) >>= (@?= Nothing)++explain :: Text -> Session Value+explain sql = statement () $ unpreparable ("explain (analyze, buffers, format json) " <> sql) mempty (D.singleRow (D.column (D.nonNullable D.json)))++scalarBool :: Text -> Session Bool+scalarBool sql = statement () (unpreparable sql mempty (D.singleRow (D.column (D.nonNullable D.bool))))++scalarText :: Text -> Session Text+scalarText sql = statement () (unpreparable sql mempty (D.singleRow (D.column (D.nonNullable D.text))))

@@ -0,0 +1,119 @@+{-# LANGUAGE OverloadedStrings #-}++-- | IR-1 acceptance 5: every inspection read works for queues whose names+-- 'parseQueueName' rejects, because an inspection surface sees whatever exists+-- in the database, not only what this library created.+--+-- The names are created and fed through raw SQL (the Haskell API rightly+-- refuses them) and read back through the new sessions. Runs on a dedicated+-- PostgreSQL instance: a mixed-case row in pgmq.meta poisons the typed+-- listQueues decoding for every test sharing a database.+module InspectionForeignNameSpec (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 EphemeralDb (ephemeralConfig)+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.Hasql.Sessions qualified as Sessions+import Pgmq.Hasql.Statements.Types (LookupMessage (LookupMessage), PeekMessages (PeekMessages), QueueMetrics (..))+import Pgmq.Migration qualified as Migration+import Pgmq.Types (ArchivedMessage (..), Message (..), MessageId (..), parseQueueName)+import System.Random (randomRIO)+import Test.Tasty (TestTree, testGroup, withResource)+import Test.Tasty.HUnit (assertEqual, assertFailure, testCase, (@?=))++tests :: TestTree+tests =+  withResource acquireDb releaseDb $ \getDb ->+    testGroup+      "Inspection of foreign queue names"+      [ testForeignName getDb "mixed-case" ("MyQueue_" <>),+        testForeignName getDb "hyphenated" ("odd-name_" <>)+      ]++testForeignName :: IO (Pg.Database, Pool.Pool) -> String -> (Text -> Text) -> TestTree+testForeignName getDb label mkName = testCase (label <> " name: peek, lookup, archive, and metrics work") $ do+  (_, pool) <- getDb+  suffix <- T.pack . show <$> randomRIO (10000 :: Word32, 99999)+  let name = mkName suffix+  -- The whole point: this library's validator refuses the name.+  case parseQueueName name of+    Left _ -> pure ()+    Right _ -> assertFailure ("expected parseQueueName to reject " <> T.unpack name)+  assertSession pool (rawUnit ("select pgmq.create('" <> name <> "')"))+  ids <- mapM (\i -> MessageId <$> assertSession pool (rawId ("select pgmq.send('" <> name <> "', '{\"i\":" <> T.pack (show i) <> "}'::jsonb)"))) [1 .. 3 :: Int]+  firstId <- case ids of+    x : _ -> pure x+    [] -> assertFailure "expected three ids"+  page <- assertSession pool (Sessions.peekMessages (PeekMessages name Nothing 10))+  map messageId (V.toList page) @?= ids+  one <- assertSession pool (Sessions.lookupMessage (LookupMessage name firstId))+  fmap messageId one @?= Just firstId+  metrics <- assertSession pool (Sessions.queueMetricsUnvalidated name)+  queueLength metrics @?= 3+  assertEqual "metrics report the name as given" name (queueName metrics)+  void $ assertSession pool (rawBool ("select pgmq.archive('" <> name <> "', " <> T.pack (show (unMessageId firstId)) <> "::bigint)"))+  archived <- assertSession pool (Sessions.peekArchivedMessages (PeekMessages name Nothing 10))+  map (messageId . archivedMessage) (V.toList archived) @?= [firstId]+  found <- assertSession pool (Sessions.lookupArchivedMessage (LookupMessage name firstId))+  fmap (messageId . archivedMessage) found @?= Just firstId+  remaining <- assertSession pool (Sessions.peekMessages (PeekMessages name Nothing 10))+  map messageId (V.toList remaining) @?= drop 1 ids+  void $ assertSession pool (rawBool ("select pgmq.drop_queue('" <> name <> "')"))++-- Dedicated database plumbing (as in MixedCaseRemediationSpec) -----------------++acquireDb :: IO (Pg.Database, Pool.Pool)+acquireDb = do+  config <- ephemeralConfig+  started <- Pg.startCached config 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. Names are spliced into SQL text because these tests+-- must construct names the Haskell API refuses; every spliced value is built+-- above from [A-Za-z0-9_-], so splicing is safe here.++rawUnit :: Text -> Session ()+rawUnit sqlText = statement () (unpreparable sqlText mempty D.noResult)++rawBool :: Text -> Session Bool+rawBool sqlText = statement () (unpreparable sqlText mempty (D.singleRow (D.column (D.nonNullable D.bool))))++rawId :: Text -> Session Int64+rawId sqlText = statement () (unpreparable sqlText mempty (D.singleRow (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

@@ -0,0 +1,364 @@+{-# LANGUAGE OverloadedStrings #-}++-- | IR-1: reads that observe a queue or archive without leasing anything.+--+-- The non-destructive guarantee is asserted two ways: the raw cells every+-- lease touches (vt, read_ct, last_read_at) are byte-identical before and+-- after a sequence of peeks, and a consumer polling the queue concurrently+-- with a loop of peeks still receives every message exactly once with+-- read_ct = 1. Paging is keyset-only: the statements' SQL contains no OFFSET,+-- and paging by the last id seen visits every message in a quiescent fixture+-- exactly once. Late commits and late archival require a first-page refresh.+module InspectionSpec (tests) where++import Control.Concurrent (forkFinally, killThread, threadDelay)+import Control.Concurrent.MVar (newEmptyMVar, putMVar, takeMVar)+import Control.Exception (IOException, bracket, finally, mask, throwIO, try)+import Control.Monad (forM_, replicateM_, void)+import Data.Aeson (object, (.=))+import Data.Int (Int32, Int64)+import Data.List (sort)+import Data.Set qualified as Set+import Data.Text (Text)+import Data.Text qualified as T+import Data.Time (UTCTime)+import Data.Vector qualified as V+import EphemeralDb (Database, TestFixture (..), withTestFixture)+import EphemeralPg qualified as Pg+import Hasql.Connection qualified as Connection+import Hasql.Decoders qualified as D+import Hasql.Encoders qualified as E+import Hasql.Pool qualified as Pool+import Hasql.Session (Session, statement)+import Hasql.Session qualified as Session+import Hasql.Statement (preparable, toSql, unpreparable)+import Pgmq.Hasql.Sessions qualified as Sessions+import Pgmq.Hasql.Statements.Inspection qualified as Inspect+import Pgmq.Hasql.Statements.Types+  ( BatchMessageQuery (..),+    BatchSendMessage (..),+    CreatePartitionedQueue (..),+    LookupMessage (LookupMessage),+    PeekMessages (PeekMessages),+    QueueMetrics (..),+    ReadMessage (..),+    SendMessage (..),+  )+import Pgmq.Types (ArchivedMessage (..), Message (..), MessageBody (..), MessageId (..), QueueName, queueNameToText)+import System.Environment (lookupEnv)+import System.Timeout (timeout)+import Test.Tasty (TestTree, testGroup)+import Test.Tasty.HUnit (assertBool, assertEqual, assertFailure, testCase, (@?=))+import TestUtils (assertSession, cleanupQueue)++tests :: Pool.Pool -> Database -> TestTree+tests p db =+  testGroup+    "Inspection (non-destructive reads)"+    [ testPeekLeavesRowsUntouched p,+      testPeekDoesNotDisturbConsumer p,+      testConsumerFailure,+      testLateCommit p db,+      testLateArchive p,+      testKeysetPagingVisitsEachOnce p,+      testNoOffset,+      testArchiveReads p,+      testLookups p,+      testLimitSemantics p,+      testMissingQueue p,+      testMetricsParity p,+      testPartitionedQueue p+    ]++-- | The cells a lease would change, for every row, in id order.+rowImages :: Text -> Session [(Int64, Int32, Maybe UTCTime, UTCTime)]+rowImages table = statement () (unpreparable sql mempty decoder)+  where+    -- Test queue names are generated from [a-z0-9_], so splicing is safe here.+    sql = "select msg_id, read_ct, last_read_at, vt from pgmq.q_" <> table <> " order by msg_id"+    decoder =+      D.rowList+        ( (,,,)+            <$> D.column (D.nonNullable D.int8)+            <*> D.column (D.nonNullable D.int4)+            <*> D.column (D.nullable D.timestamptz)+            <*> D.column (D.nonNullable D.timestamptz)+        )++sendN :: Pool.Pool -> QueueName -> Int -> IO [MessageId]+sendN pool q n =+  assertSession pool $+    Sessions.batchSendMessage+      BatchSendMessage+        { queueName = q,+          messageBodies = [MessageBody (object ["n" .= i]) | i <- [1 .. n]],+          delay = Nothing+        }++peek :: Pool.Pool -> QueueName -> Maybe MessageId -> Int32 -> IO [Message]+peek pool q cursor n =+  V.toList <$> assertSession pool (Sessions.peekMessages (PeekMessages (queueNameToText q) cursor n))++lastId :: [Message] -> IO MessageId+lastId [] = assertFailure "expected a non-empty page"+lastId page = pure (messageId (last page))++testPeekLeavesRowsUntouched :: Pool.Pool -> TestTree+testPeekLeavesRowsUntouched p = testCase "peek leaves vt, read_ct, and last_read_at byte-identical" $+  withTestFixture p $ \TestFixture {pool, queueName} -> flip finally (cleanupQueue pool queueName) $ do+    assertSession pool (Sessions.createQueue queueName)+    _ <- sendN pool queueName 3+    -- Two more with a delay, so vt differs between rows.+    _ <- assertSession pool (Sessions.sendMessage (SendMessage queueName (MessageBody "late") (Just 60)))+    _ <- assertSession pool (Sessions.sendMessage (SendMessage queueName (MessageBody "later") (Just 120)))+    -- Lease one through the real read so a non-trivial read_ct and last_read_at exist.+    leased <- assertSession pool (Sessions.readMessage (ReadMessage queueName 30 (Just 1) Nothing))+    V.length leased @?= 1+    before <- assertSession pool (rowImages (queueNameToText queueName))+    replicateM_ 5 $ do+      first <- peek pool queueName Nothing 2+      cursor <- lastId first+      void (peek pool queueName (Just cursor) 50)+    _ <- assertSession pool (Sessions.lookupMessage (LookupMessage (queueNameToText queueName) (messageId (V.head leased))))+    after <- assertSession pool (rowImages (queueNameToText queueName))+    assertEqual "rows are identical after peeks and lookups" before after+    length before @?= 5++testPeekDoesNotDisturbConsumer :: Pool.Pool -> TestTree+testPeekDoesNotDisturbConsumer p = testCase "a concurrently polling consumer sees every message exactly once" $+  withTestFixture p $ \TestFixture {pool, queueName} -> flip finally (cleanupQueue pool queueName) $ do+    assertSession pool (Sessions.createQueue queueName)+    sent <- sendN pool queueName 50+    -- The bracketed worker propagates failures and is cancelled if peeking fails.+    let consume seen empties+          | Set.size seen >= 50 || empties >= (200 :: Int) = pure seen+          | otherwise = do+              batch <- assertSession pool (Sessions.readMessage (ReadMessage queueName 60 (Just 5) Nothing))+              if V.null batch+                then threadDelay 5_000 >> consume seen (empties + 1)+                else consume (foldr (Set.insert . messageId) seen (V.toList batch)) empties+    seen <-+      withWorker (consume Set.empty 0) $+        replicateM_ 200 (void (peek pool queueName Nothing 50))+    assertEqual "the consumer received exactly the fifty sent ids" (Set.fromList sent) seen+    rows <- assertSession pool (rowImages (queueNameToText queueName))+    length rows @?= 50+    forM_ rows $ \(mid, readCt, _, _) ->+      assertEqual ("read_ct of " <> show mid <> " was bumped only by the consumer") 1 readCt+    -- Everything is leased for sixty seconds now; a further read returns nothing,+    -- which a destructive "peek" would have made impossible to predict.+    more <- assertSession pool (Sessions.readMessage (ReadMessage queueName 60 (Just 50) Nothing))+    V.length more @?= 0++-- | Base-only structured concurrency: every exit cancels the worker, and its+-- exception is rethrown in the parent rather than leaving an empty result MVar.+withWorker :: IO a -> IO () -> IO a+withWorker work during = mask $ \restore -> do+  result <- newEmptyMVar+  bracket+    (forkFinally (restore work) (putMVar result))+    killThread+    (\_ -> restore (during >> takeMVar result >>= either throwIO pure))++testConsumerFailure :: TestTree+testConsumerFailure = testCase "consumer exceptions reach the parent promptly" $ do+  outcome <-+    timeout 1_000_000 $+      try @IOException (withWorker (ioError (userError "forced consumer failure")) (pure ()))+  case outcome of+    Just (Left err) -> assertBool "original worker failure" ("forced consumer failure" `T.isInfixOf` T.pack (show err))+    Just (Right ()) -> assertFailure "worker failure was swallowed"+    Nothing -> assertFailure "parent hung waiting for a failed consumer"++testLateCommit :: Pool.Pool -> Database -> TestTree+testLateCommit p db = testCase "a late lower-id commit needs a first-page refresh" $+  withTestFixture p $ \TestFixture {pool, queueName} -> flip finally (cleanupQueue pool queueName) $ do+    assertSession pool (Sessions.createQueue queueName)+    -- Hold the first producer transaction on its own connection. The pool's+    -- second producer commits before the first, without timing-based sleeps.+    bracket+      (Connection.acquire (Pg.connectionSettings db) >>= either (assertFailure . show) pure)+      Connection.release+      $ \producer -> do+        let run :: Session a -> IO a+            run session = Connection.use producer session >>= either (assertFailure . show) pure+        run (Session.script "BEGIN")+        older <- run (Sessions.sendMessage (SendMessage queueName (MessageBody "older") Nothing))+        newer <- assertSession pool (Sessions.sendMessage (SendMessage queueName (MessageBody "newer") Nothing))+        assertBool "allocation order differs from commit order" (older < newer)+        first <- peek pool queueName Nothing 10+        map messageId first @?= [newer]+        run (Session.script "COMMIT")+        continued <- peek pool queueName (Just newer) 10+        map messageId continued @?= []+        refreshed <- peek pool queueName Nothing 10+        map messageId refreshed @?= [older, newer]++testLateArchive :: Pool.Pool -> TestTree+testLateArchive p = testCase "archiving an older id needs a first-page archive refresh" $+  withTestFixture p $ \TestFixture {pool, queueName} -> flip finally (cleanupQueue pool queueName) $ do+    assertSession pool (Sessions.createQueue queueName)+    ids <- sendN pool queueName 2+    (older, newer) <- case ids of+      [a, b] -> pure (a, b)+      other -> assertFailure ("expected two ids: " <> show other)+    let q = queueNameToText queueName+        archive mid = void (assertSession pool (Sessions.batchArchiveMessages (BatchMessageQuery queueName [mid])))+        page cursor = map (messageId . archivedMessage) . V.toList <$> assertSession pool (Sessions.peekArchivedMessages (PeekMessages q cursor 10))+    archive newer+    page Nothing >>= (@?= [newer])+    archive older+    page (Just newer) >>= (@?= [])+    page Nothing >>= (@?= [older, newer])++testKeysetPagingVisitsEachOnce :: Pool.Pool -> TestTree+testKeysetPagingVisitsEachOnce p = testCase "paging by the last id visits every message exactly once" $+  withTestFixture p $ \TestFixture {pool, queueName} -> flip finally (cleanupQueue pool queueName) $ do+    assertSession pool (Sessions.createQueue queueName)+    sent <- concat <$> mapM (const (sendN pool queueName 100)) [1 .. 10 :: Int]+    length sent @?= 1000+    before <- assertSession pool (rowImages (queueNameToText queueName))+    let go cursor pages acc = do+          page <- peek pool queueName cursor 7+          let acc' = acc <> map messageId page+          if length page < 7+            then pure (pages + 1 :: Int, acc')+            else do+              next <- lastId page+              go (Just next) (pages + 1) acc'+    (pages, visited) <- go Nothing 0 []+    assertEqual "143 pages of at most seven" 143 pages+    assertEqual "every id exactly once, in order" (sort sent) visited+    assertEqual "no duplicates" (length visited) (Set.size (Set.fromList visited))+    after <- assertSession pool (rowImages (queueNameToText queueName))+    assertEqual "paging modified nothing" before after++testNoOffset :: TestTree+testNoOffset = testCase "inspection statements never use OFFSET" $ do+  let sqls =+        [ toSql (Inspect.peekStatement "q_x"),+          toSql (Inspect.peekArchivedStatement "a_x"),+          toSql (Inspect.lookupStatement "q_x"),+          toSql (Inspect.lookupArchivedStatement "a_x")+        ]+  forM_ sqls $ \sql -> do+    assertBool ("no offset in " <> T.unpack sql) (not ("offset" `T.isInfixOf` T.toLower sql))+    assertBool ("quoted identifier in " <> T.unpack sql) ("pgmq.\"" `T.isInfixOf` sql)+  assertBool "pages are ordered by msg_id" ("order by msg_id asc limit $2" `T.isInfixOf` toSql (Inspect.peekStatement "q_x"))+  Inspect.quoteIdentifier "q_odd-name" @?= "\"q_odd-name\""+  Inspect.quoteIdentifier "we\"ird" @?= "\"we\"\"ird\""++testArchiveReads :: Pool.Pool -> TestTree+testArchiveReads p = testCase "archive reads return archived messages with their archival timestamp" $+  withTestFixture p $ \TestFixture {pool, queueName} -> flip finally (cleanupQueue pool queueName) $ do+    assertSession pool (Sessions.createQueue queueName)+    sent <- sendN pool queueName 3+    archived <- assertSession pool (Sessions.batchArchiveMessages (BatchMessageQuery queueName (take 2 sent)))+    sort archived @?= sort (take 2 sent)+    let q = queueNameToText queueName+    rows <- V.toList <$> assertSession pool (Sessions.peekArchivedMessages (PeekMessages q Nothing 10))+    map (messageId . archivedMessage) rows @?= take 2 sent+    forM_ rows $ \row ->+      assertBool "archived_at is not before enqueued_at" (archivedAt row >= enqueuedAt (archivedMessage row))+    live <- peek pool queueName Nothing 10+    map messageId live @?= drop 2 sent+    -- The archive pages by the same exclusive cursor.+    second <- V.toList <$> assertSession pool (Sessions.peekArchivedMessages (PeekMessages q (Just (sent !! 0)) 10))+    map (messageId . archivedMessage) second @?= [sent !! 1]++testLookups :: Pool.Pool -> TestTree+testLookups p = testCase "lookup finds a present message and answers Nothing for an absent one, in both tables" $+  withTestFixture p $ \TestFixture {pool, queueName} -> flip finally (cleanupQueue pool queueName) $ do+    assertSession pool (Sessions.createQueue queueName)+    sent <- sendN pool queueName 2+    (liveId, archivedId) <- case sent of+      [a, b] -> pure (a, b)+      other -> assertFailure ("expected two ids, got " <> show other)+    _ <- assertSession pool (Sessions.batchArchiveMessages (BatchMessageQuery queueName [archivedId]))+    let q = queueNameToText queueName+    live <- assertSession pool (Sessions.lookupMessage (LookupMessage q liveId))+    fmap messageId live @?= Just liveId+    gone <- assertSession pool (Sessions.lookupMessage (LookupMessage q archivedId))+    fmap messageId gone @?= Nothing+    inArchive <- assertSession pool (Sessions.lookupArchivedMessage (LookupMessage q archivedId))+    fmap (messageId . archivedMessage) inArchive @?= Just archivedId+    notInArchive <- assertSession pool (Sessions.lookupArchivedMessage (LookupMessage q liveId))+    fmap (messageId . archivedMessage) notInArchive @?= Nothing+    never <- assertSession pool (Sessions.lookupMessage (LookupMessage q (MessageId 999_999_999)))+    fmap messageId never @?= Nothing++testLimitSemantics :: Pool.Pool -> TestTree+testLimitSemantics p = testCase "limit bounds the page and a cursor excludes its own row" $+  withTestFixture p $ \TestFixture {pool, queueName} -> flip finally (cleanupQueue pool queueName) $ do+    assertSession pool (Sessions.createQueue queueName)+    sent <- sendN pool queueName 5+    three <- peek pool queueName Nothing 3+    map messageId three @?= take 3 sent+    rest <- peek pool queueName (Just (sent !! 2)) 3+    map messageId rest @?= drop 3 sent+    none <- peek pool queueName (Just (sent !! 4)) 3+    map messageId none @?= []+    zero <- peek pool queueName Nothing 0+    map messageId zero @?= []++testMissingQueue :: Pool.Pool -> TestTree+testMissingQueue p = testCase "a missing queue fails with undefined_table (42P01)" $ do+  result <- Pool.use p (Sessions.peekMessages (PeekMessages "no_such_queue_for_inspection" Nothing 1))+  case result of+    Right _ -> assertFailure "peeking a missing queue should fail"+    Left err -> assertBool ("expected 42P01, got " <> show err) ("42P01" `T.isInfixOf` T.pack (show err))++testMetricsParity :: Pool.Pool -> TestTree+testMetricsParity p = testCase "queueMetricsUnvalidated matches queueMetrics for a validated name" $+  withTestFixture p $ \TestFixture {pool, queueName} -> flip finally (cleanupQueue pool queueName) $ do+    assertSession pool (Sessions.createQueue queueName)+    _ <- sendN pool queueName 4+    typed <- assertSession pool (Sessions.queueMetrics queueName)+    lenient <- assertSession pool (Sessions.queueMetricsUnvalidated (queueNameToText queueName))+    queueLength lenient @?= queueLength typed+    queueVisibleLength lenient @?= queueVisibleLength typed+    totalMessages lenient @?= totalMessages typed+    defaultPartitionLength lenient @?= defaultPartitionLength typed+    queueLength lenient @?= 4++-- | A partitioned queue's physical table is a partitioned parent; a SELECT on+-- it sees every partition, so the reads need no special case. Needs pg_partman.+testPartitionedQueue :: Pool.Pool -> TestTree+testPartitionedQueue p = testCase "peek and lookup work on a partitioned queue (needs pg_partman)" $+  withPartman p $+    withTestFixture p $ \TestFixture {pool, queueName} -> flip finally (cleanupQueue pool queueName) $ do+      assertSession pool (Sessions.createPartitionedQueue (CreatePartitionedQueue queueName "10" "100"))+      sent <- sendN pool queueName 12+      page <- peek pool queueName Nothing 20+      map messageId page @?= sent+      one <- assertSession pool (Sessions.lookupMessage (LookupMessage (queueNameToText queueName) (sent !! 11)))+      fmap messageId one @?= Just (sent !! 11)+      -- The interval-10 archive also spans multiple leaves and supports the+      -- same first/late cursor and by-id paths as the larger diagnostic fixture.+      archivedIds <- assertSession pool (Sessions.batchArchiveMessages (BatchMessageQuery queueName sent))+      sort archivedIds @?= sent+      let q = queueNameToText queueName+          archivePage cursor = map (messageId . archivedMessage) . V.toList <$> assertSession pool (Sessions.peekArchivedMessages (PeekMessages q cursor 5))+      archivePage Nothing >>= (@?= take 5 sent)+      archivePage (Just (sent !! 10)) >>= (@?= drop 11 sent)+      found <- assertSession pool (Sessions.lookupArchivedMessage (LookupMessage q (sent !! 11)))+      fmap (messageId . archivedMessage) found @?= Just (sent !! 11)+      missing <- assertSession pool (Sessions.lookupArchivedMessage (LookupMessage q (MessageId 999999)))+      fmap (messageId . archivedMessage) missing @?= Nothing++withPartman :: Pool.Pool -> IO () -> IO ()+withPartman pool action = do+  available <-+    assertSession pool $+      statement () $+        preparable+          "select exists (select 1 from pg_extension where extname = 'pg_partman')"+          E.noParams+          (D.singleRow (D.column (D.nonNullable D.bool)))+  required <- (== Just "1") <$> lookupEnv "PGMQ_REQUIRE_PARTMAN"+  if available+    then action+    else+      if required+        then assertFailure "PGMQ_REQUIRE_PARTMAN=1 but pg_partman is not installed"+        else putStrLn "    SKIPPED: pg_partman is not installed"

@@ -0,0 +1,76 @@+{-# LANGUAGE OverloadedStrings #-}++-- | Pins the wire shape of 'QueueMetrics' and proves that 'Message' survives+-- encode-then-decode for arbitrary jsonb-storable bodies and headers. The+-- property lives here rather than in pgmq-core because this suite already+-- owns the hedgehog generators PostgreSQL accepts.+module JsonGoldenSpec (tests) where++import Data.Aeson (ToJSON (..), Value (Null), decode, encode)+import Data.Time (UTCTime (..), fromGregorian)+import Data.Time.Clock.POSIX (posixSecondsToUTCTime)+import Generators (genJsonValue)+import Hedgehog (Gen, forAll, property, (===))+import Hedgehog.Gen qualified as Gen+import Hedgehog.Range qualified as Range+import Pgmq.Hasql.Statements.Types (QueueMetrics (..))+import Pgmq.Types (ArchivedMessage (..), Message (..), MessageBody (..), MessageId (..))+import Test.Tasty (TestTree, localOption, testGroup)+import Test.Tasty.Golden (NoCreateFile (..), goldenVsString)+import Test.Tasty.Hedgehog (testProperty)++tests :: TestTree+tests =+  localOption (NoCreateFile True) $+    testGroup+      "JSON wire encodings"+      [ goldenVsString "queue_metrics" "test/golden/queue_metrics.json" (pure (encode sampleMetrics <> "\n")),+        goldenVsString "queue_metrics_partitioned" "test/golden/queue_metrics_partitioned.json" (pure (encode partitionedMetrics <> "\n")),+        testProperty "Message round-trips modulo explicit-null header normalization" $ property $ do+          m <- forAll genMessage+          let normalized = m {headers = case headers m of Just Null -> Nothing; other -> other}+          decode (encode m) === Just normalized+          encode normalized === encode m+          decode (encode m) === Just (toJSON m)+          at <- forAll genTime+          let archived = ArchivedMessage m at+          decode (encode archived) === Just (ArchivedMessage normalized at)+          decode (encode archived) === Just (toJSON archived)+          encode (ArchivedMessage normalized at) === encode archived,+        testProperty "QueueMetrics encoder parity" $ property $ do+          newest <- forAll (Gen.maybe (Gen.int32 (Range.linear 0 maxBound)))+          oldest <- forAll (Gen.maybe (Gen.int32 (Range.linear 0 maxBound)))+          estimate <- forAll (Gen.maybe (Gen.int64 (Range.linear 0 maxBound)))+          let m = QueueMetrics "orders" 120 newest oldest 9000 scrape 115 estimate+          decode (encode m) === Just (toJSON m)+      ]++scrape :: UTCTime+scrape = UTCTime (fromGregorian 2026 8 19) 45296.123++-- | An empty queue: both ages are null, the partition estimate is null.+sampleMetrics :: QueueMetrics+sampleMetrics = QueueMetrics "orders" 0 Nothing Nothing 0 scrape 0 Nothing++-- | A partitioned queue on PGMQ 1.13 with a planner estimate.+partitionedMetrics :: QueueMetrics+partitionedMetrics = QueueMetrics "events" 120 (Just 2) (Just 3600) 9000 scrape 115 (Just 7)++-- | Whole milliseconds so the timestamp survives jsonb-independent rendering+-- exactly; aeson itself round-trips picoseconds, but the property is about+-- the record, not the clock.+genTime :: Gen UTCTime+genTime = do+  millis <- Gen.integral (Range.linear 0 (4_102_444_800_000 :: Integer))+  pure (posixSecondsToUTCTime (fromIntegral millis / 1000))++genMessage :: Gen Message+genMessage =+  Message+    <$> (MessageId <$> Gen.integral (Range.linear 1 maxBound))+    <*> genTime+    <*> genTime+    <*> Gen.maybe genTime+    <*> Gen.integral (Range.linear 0 1000)+    <*> (MessageBody <$> genJsonValue)+    <*> Gen.maybe genJsonValue

@@ -6,7 +6,11 @@ import AliasingSpec qualified import AllFunctionsDecoderSpec qualified import DecoderValidationSpec qualified-import EphemeralDb (withPgmqDb)+import EphemeralDb (ensureOwnerOnlyRoot, withPgmqDb)+import InspectionCostSpec qualified+import InspectionForeignNameSpec qualified+import InspectionSpec qualified+import JsonGoldenSpec qualified import MessageSpec qualified import MetricsSpec qualified import MixedCaseRemediationSpec qualified@@ -14,7 +18,9 @@ import NotifyRaceSpec qualified import NullBodySpec qualified import NullSemanticsSpec qualified+import PartitionedNotifySpec qualified import QueueSpec qualified+import RootPermissionsSpec qualified import RoundTripSpec qualified import SchemaSpec qualified import System.Environment (lookupEnv)@@ -30,8 +36,12 @@     let tree =           testGroup             "pgmq-hasql"-            ( [ QueueSpec.tests pool,+            ( [ RootPermissionsSpec.tests ensureOwnerOnlyRoot,+                JsonGoldenSpec.tests,+                QueueSpec.tests pool,                 MessageSpec.tests pool,+                InspectionSpec.tests pool db,+                InspectionCostSpec.tests pool,                 MetricsSpec.tests,                 AdvancedOpsSpec.tests pool,                 NullSemanticsSpec.tests pool,@@ -43,9 +53,14 @@                 -- sweeps every mixed-case row in its database.                 AliasingSpec.tests,                 MixedCaseRemediationSpec.tests,+                -- Creates mixed-case (and hyphenated) queues through raw SQL to+                -- prove the inspection reads accept foreign names, so it too+                -- runs on its own instance.+                InspectionForeignNameSpec.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,+                PartitionedNotifySpec.tests,                 SchemaSpec.tests pool,                 RoundTripSpec.tests pool,                 DecoderValidationSpec.tests pool,

@@ -10,7 +10,7 @@ -- 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+module NotifyChannelSpec (tests, withListener, awaitNotify) where  import Control.Concurrent (threadDelay) import Control.Exception (bracket)@@ -120,7 +120,9 @@ awaitNotify conn attempts   | attempts <= 0 = pure Nothing   | otherwise = do-      _ <- LibPQ.consumeInput conn+      ok <- LibPQ.consumeInput conn+      unless ok (assertFailure "libpq consumeInput failed")+      LibPQ.status conn >>= (@?= LibPQ.ConnectionOk)       pending <- LibPQ.notifies conn       case pending of         Just n -> pure (Just n)

@@ -0,0 +1,101 @@+{-# LANGUAGE OverloadedStrings #-}++-- | Partition silence, including native metadata aliases, with a live control.+module PartitionedNotifySpec (tests) where++import Control.Exception (finally)+import Control.Monad (forM_, unless)+import Data.Text (Text)+import Data.Text qualified as T+import Data.Text.Encoding qualified as TE+import Database.PostgreSQL.LibPQ qualified as PQ+import EphemeralDb (withPgmqDb)+import Hasql.Pool qualified as Pool+import Hasql.Session qualified as Session+import NotifyChannelSpec (awaitNotify, withListener)+import System.Environment (lookupEnv)+import Test.Tasty (TestTree, testGroup)+import Test.Tasty.HUnit (assertFailure, testCase, (@?=))++tests :: TestTree+tests =+  testGroup+    "PartitionedNotifySpec"+    [ testCase "partition inserts are silent and leave throttle at epoch" (partitionCase False False),+      testCase "native leaf metadata alias stays silent" (nativeOnly (partitionCase True False)),+      testCase "native leaf throttle alias stays silent and unchanged" (nativeOnly (partitionCase True True)),+      testCase "ordinary nonzero throttle suppresses second committed insert" ordinaryCase,+      testCase "native mixed-case missing lowercase metadata stays silent" (nativeOnly mixedCase)+    ]++nativeOnly :: IO () -> IO ()+nativeOnly action = do+  stock <- (== Just "1.12.0") <$> lookupEnv "PGMQ_TEST_SCHEMA_VERSION"+  if stock then putStrLn "SKIPPED: native alias hardening" else action++isolated :: (Pool.Pool -> (Text -> (PQ.Connection -> IO ()) -> IO ()) -> IO ()) -> IO ()+isolated action = do+  result <- withPgmqDb $ \pool db -> action pool (withListener db) `finally` Pool.release pool+  either (assertFailure . show) pure result++sql :: Pool.Pool -> Text -> IO ()+sql pool command = Pool.use pool (Session.script command) >>= either (assertFailure . show) pure++-- A separate listener subscription for each leaf, on the same live connection.+listen :: PQ.Connection -> Text -> IO ()+listen conn channel = do+  result <- PQ.exec conn (TE.encodeUtf8 ("LISTEN \"" <> T.replace "\"" "\"\"" channel <> "\""))+  case result of+    Nothing -> assertFailure "LISTEN returned no result"+    Just res -> PQ.resultStatus res >>= (@?= PQ.CommandOk)++partitionCase :: Bool -> Bool -> IO ()+partitionCase alias throttle = isolated $ \pool listener -> do+  available <- Pool.use pool (Session.script "DO $$ BEGIN IF NOT EXISTS (SELECT FROM pg_extension WHERE extname='pg_partman') THEN RAISE EXCEPTION 'pg_partman unavailable'; END IF; END $$")+  required <- (== Just "1") <$> lookupEnv "PGMQ_REQUIRE_PARTMAN"+  case available of+    Left err -> if required then assertFailure (show err) else putStrLn "SKIPPED: pg_partman unavailable"+    Right () -> do+      sql pool "SELECT pgmq.create_partitioned('pq','10','100'); SELECT pgmq.enable_notify_insert('pq',0); SELECT pgmq.create('control'); SELECT pgmq.enable_notify_insert('control',0)"+      if alias then sql pool "SELECT pgmq.create('pq_p0'); DO $$ BEGIN IF NOT (SELECT relispartition FROM pg_class WHERE oid='pgmq.q_pq_p0'::regclass) THEN RAISE EXCEPTION 'alias changed relation'; END IF; END $$" else pure ()+      if throttle then sql pool "INSERT INTO pgmq.notify_insert_throttle(queue_name,throttle_interval_ms,last_notified_at) VALUES ('pq_p0',0,to_timestamp(0))" else pure ()+      listener "pgmq.q_control.INSERT" $ \conn -> do+        result <- PQ.exec conn "SELECT 'pgmq.' || c.relname || '.INSERT' FROM pg_inherits i JOIN pg_class c ON c.oid=i.inhrelid WHERE i.inhparent='pgmq.q_pq'::regclass"+        case result of+          Nothing -> assertFailure "partition query failed"+          Just res -> do+            PQ.resultStatus res >>= (@?= PQ.TuplesOk)+            rows <- PQ.ntuples res+            forM_ [0 .. rows - 1] $ \row -> do+              channel <- PQ.getvalue res row 0+              maybe (assertFailure "null partition channel") (listen conn . TE.decodeUtf8) channel+        listen conn "pgmq.q_pq.INSERT"+        sql pool "SELECT pgmq.send('control','{}')"+        control <- awaitNotify conn 20+        fmap PQ.notifyRelname control @?= Just "pgmq.q_control.INSERT"+        forM_ [1 :: Int .. 3] $ \_ -> sql pool "SELECT pgmq.send('pq','{}')"+        received <- awaitNotify conn 10+        fmap PQ.notifyRelname received @?= Nothing+      sql pool "DO $$ BEGIN IF (SELECT count(*) FROM pgmq.q_pq) <> 3 THEN RAISE EXCEPTION 'messages not stored'; END IF; IF EXISTS (SELECT FROM pgmq.notify_insert_throttle WHERE queue_name IN ('pq','pq_p0') AND last_notified_at <> to_timestamp(0)) THEN RAISE EXCEPTION 'partition throttle advanced'; END IF; END $$"++ordinaryCase :: IO ()+ordinaryCase = isolated $ \pool listener -> do+  sql pool "SELECT pgmq.create('ordinary'); SELECT pgmq.enable_notify_insert('ordinary',3600000)"+  listener "pgmq.q_ordinary.INSERT" $ \conn -> do+    sql pool "SELECT pgmq.send('ordinary','{}')"+    first <- awaitNotify conn 20+    fmap PQ.notifyRelname first @?= Just "pgmq.q_ordinary.INSERT"+    sql pool "CREATE TABLE public.throttle_before AS SELECT last_notified_at FROM pgmq.notify_insert_throttle WHERE queue_name='ordinary'"+    sql pool "SELECT pgmq.send('ordinary','{}')"+    second <- awaitNotify conn 10+    fmap PQ.notifyRelname second @?= Nothing+    sql pool "DO $$ BEGIN IF (SELECT last_notified_at FROM pgmq.notify_insert_throttle WHERE queue_name='ordinary') <> (SELECT last_notified_at FROM public.throttle_before) THEN RAISE EXCEPTION 'suppressed insert advanced timestamp'; END IF; END $$"+    sql pool "DO $$ BEGIN IF (SELECT count(*) FROM pgmq.q_ordinary) <> 2 THEN RAISE EXCEPTION 'messages not stored'; END IF; END $$"++mixedCase :: IO ()+mixedCase = isolated $ \pool listener -> do+  sql pool "SELECT pgmq.create('MixedNotify'); SELECT pgmq.enable_notify_insert('MixedNotify',0)"+  listener "pgmq.q_mixednotify.INSERT" $ \conn -> do+    sql pool "SELECT pgmq.send('MixedNotify','{}')"+    received <- awaitNotify conn 10+    unless (fmap PQ.notifyRelname received == Nothing) (assertFailure "mixed-case queue notified without matching metadata")

@@ -0,0 +1,63 @@+-- | Check each suite's local root helper using private disposable fixtures.+module RootPermissionsSpec (tests) where++import Control.Exception (IOException, bracket, try)+import Data.Bits ((.&.))+import System.Directory (getTemporaryDirectory, removePathForcibly)+import System.Posix.Directory qualified as Posix+import System.Posix.Files (createSymbolicLink, fileMode, getSymbolicLinkStatus, isDirectory, setFileMode)+import System.Posix.Temp (mkdtemp)+import Test.Tasty (TestTree, testGroup)+import Test.Tasty.HUnit (assertBool, assertFailure, testCase, (@?=))++tests :: (FilePath -> IO ()) -> TestTree+tests ensure =+  testGroup+    "RootPermissionsSpec"+    [ testCase "missing root is a real 0700 directory" $ private $ \parent -> do+        let root = parent <> "/root"+        ensure root+        status <- getSymbolicLinkStatus root+        assertBool "real directory" (isDirectory status)+        fileMode status .&. 0o7777 @?= 0o700,+      testCase "owned 0755 directory is tightened without losing contents" $ private $ \parent -> do+        let root = parent <> "/root"+        Posix.createDirectory root 0o755+        setFileMode root 0o755+        writeFile (root <> "/sentinel") "preserve"+        ensure root+        mode root >>= (@?= 0o700)+        readFile (root <> "/sentinel") >>= (@?= "preserve"),+      testCase "repeated call preserves contents and permissions" $ private $ \parent -> do+        let root = parent <> "/root"+        ensure root+        writeFile (root <> "/sentinel") "preserve"+        ensure root+        mode root >>= (@?= 0o700)+        readFile (root <> "/sentinel") >>= (@?= "preserve"),+      testCase "regular file is rejected unchanged" $ private $ \parent -> do+        let root = parent <> "/root"+        writeFile root "preserve"+        setFileMode root 0o644+        rejected (ensure root)+        mode root >>= (@?= 0o644)+        readFile root >>= (@?= "preserve"),+      testCase "symlink is rejected without chmod of its target" $ private $ \parent -> do+        let root = parent <> "/root"+            target = parent <> "/target"+        Posix.createDirectory target 0o755+        setFileMode target 0o755+        writeFile (target <> "/sentinel") "preserve"+        createSymbolicLink target root+        rejected (ensure root)+        mode target >>= (@?= 0o755)+        readFile (target <> "/sentinel") >>= (@?= "preserve")+    ]+  where+    mode path = (.&. 0o7777) . fileMode <$> getSymbolicLinkStatus path+    rejected action = do+      result <- try action :: IO (Either IOException ())+      case result of+        Left _ -> pure ()+        Right () -> assertFailure "unsafe root accepted"+    private action = bracket (getTemporaryDirectory >>= \tmp -> mkdtemp (tmp <> "/pgmq-root-test-")) removePathForcibly action

@@ -4,9 +4,10 @@ {-# LANGUAGE TypeOperators #-}  -- Compile witnesses: this module intentionally uses only the public umbrella.-module UmbrellaExportsSpec (grouped, polling, creation, arguments, metric) where+module UmbrellaExportsSpec (grouped, polling, creation, arguments, metric, inspection, lenientMetrics, inspectionArguments) where  import Data.Int (Int32, Int64)+import Data.Text (Text) import Data.Vector (Vector) import Hasql.Session (Session) import Pgmq@@ -25,3 +26,12 @@  metric :: QueueMetrics -> Maybe Int64 metric = defaultPartitionLength++inspection :: PeekMessages -> LookupMessage -> (Session (Vector Message), Session (Vector ArchivedMessage), Session (Maybe Message), Session (Maybe ArchivedMessage))+inspection page one = (peekMessages page, peekArchivedMessages page, lookupMessage one, lookupArchivedMessage one)++lenientMetrics :: Text -> Session QueueMetrics+lenientMetrics = queueMetricsUnvalidated++inspectionArguments :: Text -> (PeekMessages, LookupMessage)+inspectionArguments q = (PeekMessages q Nothing 50, LookupMessage q (MessageId 1))

@@ -0,0 +1,1 @@+{"queue_name":"orders","queue_length":0,"newest_msg_age_sec":null,"oldest_msg_age_sec":null,"total_messages":0,"scrape_time":"2026-08-19T12:34:56.123Z","queue_visible_length":0,"default_partition_length":null}

@@ -0,0 +1,1 @@+{"queue_name":"events","queue_length":120,"newest_msg_age_sec":2,"oldest_msg_age_sec":3600,"total_messages":9000,"scrape_time":"2026-08-19T12:34:56.123Z","queue_visible_length":115,"default_partition_length":7}