keiro-pgmq 0.16.0.0 → 0.17.0.0

raw patch · 6 files changed

+892/−141 lines, 6 filesdep ~keiro-coredep ~keiro-test-supportdep ~pgmq-configPVP ok

version bump matches the API change (PVP)

Dependency ranges changed: keiro-core, keiro-test-support, pgmq-config, pgmq-core, pgmq-effectful, pgmq-hasql, pgmq-migration, shibuya-pgmq-adapter

API changes (from Hackage documentation)

Files

• CHANGELOG.md +45/−0
• keiro-pgmq.cabal +69/−60
• src/Keiro/PGMQ/Dlq.hs +103/−28
• src/Keiro/PGMQ/Job.hs +164/−34
• src/Keiro/PGMQ/Metrics.hs +3/−2
• test/Main.hs +508/−17

@@ -6,6 +6,51 @@  ## [Unreleased] +## 0.17.0.0 — 2026-09-17++### Breaking Changes++- `DlqEntry` now exposes `originalHeaders`; external record construction and+  exhaustive positional matches must account for the new field. `purgeDlq` now+  returns `PurgeDlqResult` instead of `()`. Callers that discard the result may+  still compile but must inspect `PurgeDlqBlocked` rather than assuming deletion.+- Require the released `pgmq-config`, `pgmq-core`, `pgmq-effectful`,+  `pgmq-hasql`, and test-only `pgmq-migration` 0.6 family together with+  `shibuya-pgmq-adapter ^>=0.16.0.0`. The re-exported `QueueMetrics` record now+  includes nullable `defaultPartitionLength`, so callers that construct it or+  match it positionally must handle the new field. Partitioned provisioning+  deliberately passes `premake = Nothing` to retain the PGMQ server default+  rather than exposing explicit premake control.+- `Job` gains the required `jobOrdering` field. Explicit tuning must match the+  declaration; invalid raw tuning and legacy FIFO batches larger than one now+  throw `JobConsumptionConfigError` before adapter construction or reads.++### New Features++- Add `FifoHeads`, backed by PGMQ's grouped-head read in both worker and+  one-shot paths. It safely batches independent absolute group heads while a+  failed, invisible, or delayed head blocks successors in its own group.+- Add `mkPartitionSpec` and `PartitionSpecConfigError` to reject obvious empty,+  mixed-unit, non-positive, out-of-range, and too-short numeric partition+  settings before database access. The raw constructor remains available for+  server-specific values.++- DLQ redrive preserves the wrapper's original producer headers, including FIFO,+  trace, and application metadata. Missing and JSON-null legacy headers remain+  headerless and never fall back to the DLQ row's own headers.+- `archiveDlqEntries` archives known message ids even while inspection has hidden+  them. Ordinary `purgeDlq` refuses when metrics report hidden rows and returns+  the deleted or blocked count; `purgeDlqForce` is the distinctly named+  unconditional escape hatch.++### Other Changes++- Clarify that partition retention drops whole active and archive partitions and+  can remove unprocessed work after a sufficiently long outage or backlog.+- Describe FIFO provisioning as PGMQ's conventional name-detected GIN index.+  No supplemental grouped-read index is recommended pending reproducible+  full-workload measurements, and Keiro installs no experimental startup DDL.+ ## 0.16.0.0 — 2026-09-07  ### Other Changes

@@ -1,7 +1,7 @@-cabal-version:   3.0-name:            keiro-pgmq-version:         0.16.0.0-synopsis:        PostgreSQL job-queue (PGMQ) integration for Keiro+cabal-version: 3.0+name: keiro-pgmq+version: 0.17.0.0+synopsis: PostgreSQL job-queue (PGMQ) integration for Keiro description:   A typed background-job queue for Keiro applications on top of PGMQ (the   PostgreSQL-native message queue) and shibuya (a Broadway-style worker@@ -9,29 +9,34 @@   retry/dead-letter policy, then write a plain domain handler of type   @p -> Eff es JobOutcome@. -license:         BSD-3-Clause-license-file:    LICENSE-author:          Nadeem Bitar-maintainer:      nadeem@gmail.com-copyright:       2026 Nadeem Bitar-category:        Control-homepage:        https://github.com/shinzui/keiro#readme-bug-reports:     https://github.com/shinzui/keiro/issues-build-type:      Simple-tested-with:     GHC >=9.12 && <9.13+license: BSD-3-Clause+license-file: LICENSE+author: Nadeem Bitar+maintainer: nadeem@gmail.com+copyright: 2026 Nadeem Bitar+category: Control+homepage: https://github.com/shinzui/keiro#readme+bug-reports: https://github.com/shinzui/keiro/issues+build-type: Simple+tested-with: ghc >=9.12 && <9.13 extra-doc-files: CHANGELOG.md  source-repository head-  type:     git+  type: git   location: https://github.com/shinzui/keiro.git  common warnings   ghc-options:-    -Wall -Wcompat -Widentities -Wincomplete-record-updates-    -Wincomplete-uni-patterns -Wpartial-fields -Wredundant-constraints+    -Wall+    -Wcompat+    -Widentities+    -Wincomplete-record-updates+    -Wincomplete-uni-patterns+    -Wpartial-fields+    -Wredundant-constraints  common shared-  default-language:   GHC2024+  default-language: GHC2024   default-extensions:     DeriveAnyClass     DuplicateRecordFields@@ -43,7 +48,7 @@     PackageImports  library-  import:          warnings, shared+  import: warnings, shared   exposed-modules:     Keiro.PGMQ     Keiro.PGMQ.Codec@@ -52,49 +57,53 @@     Keiro.PGMQ.Metrics     Keiro.PGMQ.Runtime -  hs-source-dirs:  src+  hs-source-dirs: src   build-depends:-    , aeson                 >=2.2       && <2.3-    , base                  >=4.21      && <5-    , effectful-core        >=2.6       && <2.7-    , hasql                 >=1.10      && <1.11-    , hasql-pool            >=1.2       && <1.5-    , hs-opentelemetry-api  >=1.0       && <1.1-    , keiro-core            ^>=0.16.0.0-    , pgmq-config           >=0.5       && <0.6-    , pgmq-core             >=0.5       && <0.6-    , pgmq-effectful        >=0.5       && <0.6-    , pgmq-hasql            >=0.5       && <0.6-    , shibuya-core          ^>=0.9.0.0-    , shibuya-pgmq-adapter  ^>=0.14.0.0-    , streamly-core         >=0.3       && <0.4-    , text                  >=2.1       && <2.2-    , time                  >=1.12      && <1.15+    aeson >=2.2 && <2.3,+    base >=4.21 && <5,+    effectful-core >=2.6 && <2.7,+    hasql >=1.10 && <1.11,+    hasql-pool >=1.2 && <1.5,+    hs-opentelemetry-api >=1.0 && <1.1,+    keiro-core ^>=0.17.0.0,+    pgmq-config >=0.6 && <0.7,+    pgmq-core >=0.6 && <0.7,+    pgmq-effectful >=0.6 && <0.7,+    pgmq-hasql >=0.6 && <0.7,+    shibuya-core ^>=0.9.0.0,+    shibuya-pgmq-adapter ^>=0.16.0.0,+    streamly-core >=0.3 && <0.4,+    text >=2.1 && <2.2,+    time >=1.12 && <1.15,  test-suite keiro-pgmq-test-  import:         warnings, shared-  type:           exitcode-stdio-1.0+  import: warnings, shared+  type: exitcode-stdio-1.0   hs-source-dirs: test-  main-is:        Main.hs-  ghc-options:    -threaded -rtsopts -with-rtsopts=-N+  main-is: Main.hs+  ghc-options:+    -threaded+    -rtsopts+    -with-rtsopts=-N+   build-depends:-    , aeson                                >=2.2       && <2.3-    , base                                 >=4.21      && <5-    , effectful-core                       >=2.6       && <2.7-    , hasql                                >=1.10      && <1.11-    , hasql-pool                           >=1.2       && <1.5-    , hs-opentelemetry-api                 >=1.0       && <1.1-    , hs-opentelemetry-exporter-in-memory  >=1.0       && <1.1-    , hs-opentelemetry-propagator-w3c      >=1.0       && <1.1-    , hs-opentelemetry-sdk                 >=1.0       && <1.1-    , hspec                                >=2.11-    , keiro-core                           ^>=0.16.0.0-    , keiro-pgmq-    , keiro-test-support                   ^>=0.16.0.0-    , pgmq-config                          >=0.5       && <0.6-    , pgmq-core                            >=0.5       && <0.6-    , pgmq-effectful                       >=0.5       && <0.6-    , pgmq-migration                       >=0.5       && <0.6-    , shibuya-core                         ^>=0.9.0.0-    , shibuya-pgmq-adapter                 ^>=0.14.0.0-    , text                                 >=2.1       && <2.2+    aeson >=2.2 && <2.3,+    base >=4.21 && <5,+    effectful-core >=2.6 && <2.7,+    hasql >=1.10 && <1.11,+    hasql-pool >=1.2 && <1.5,+    hs-opentelemetry-api >=1.0 && <1.1,+    hs-opentelemetry-exporter-in-memory >=1.0 && <1.1,+    hs-opentelemetry-propagator-w3c >=1.0 && <1.1,+    hs-opentelemetry-sdk >=1.0 && <1.1,+    hspec >=2.11,+    keiro-core ^>=0.17.0.0,+    keiro-pgmq,+    keiro-test-support ^>=0.17.0.0,+    pgmq-config >=0.6 && <0.7,+    pgmq-core >=0.6 && <0.7,+    pgmq-effectful >=0.6 && <0.7,+    pgmq-migration >=0.6 && <0.7,+    shibuya-core ^>=0.9.0.0,+    shibuya-pgmq-adapter ^>=0.16.0.0,+    text >=2.1 && <2.2,

@@ -12,23 +12,46 @@ -- treat metadata as optional so operators can still inspect legacy or hand-written -- rows. ----- PGMQ does not expire DLQ rows by itself. The retention model is "archive-then---- purge": 'archiveDlq' retains dead letters by moving them into the archive table--- @pgmq.a_<dlq>@ (preserving @enqueued_at@ / @read_ct@ and stamping @archived_at@)--- for audit, while 'purgeDlq' deletes them permanently. An operator who needs an--- audit trail runs 'archiveDlq' (retain) and may then 'purgeDlq' (clear the active--- table); an operator who does not keeps using 'purgeDlq' alone. Either way, alert--- on the DLQ's depth via 'Keiro.PGMQ.Metrics.jobDlqMetrics'.+-- 'readDlq', 'redriveDlq', and count-based 'archiveDlq' read visible rows and hide+-- each row they inspect for 30 seconds. To retain inspected rows without waiting,+-- keep their 'dlqMessageId' values and pass them to 'archiveDlqEntries'. Verify+-- that every requested id was returned before considering retention complete: ----- Redrive is at-least-once: a crash after sending the original payload back to the--- main queue but before deleting the DLQ row leaves the payload in both places.--- Handlers must therefore be idempotent.+-- @+-- entries <- readDlq job 100+-- let requested = fmap dlqMessageId entries+-- archived <- archiveDlqEntries job requested+-- if Set.fromList archived /= Set.fromList requested+--   then stopAndInvestigate+--   else purgeDlq job+-- @+--+-- PGMQ does not expire ordinary DLQ rows by itself. Archiving moves active rows+-- into @pgmq.a_<dlq>@ for audit. Partitioned archive retention remains subject to+-- its configured partition maintenance; archiving does not promise indefinite+-- retention. 'purgeDlq' deletes only after a metrics snapshot reports no hidden+-- rows, returning 'PurgeDlqBlocked' otherwise. The snapshot and deletion are not+-- atomic. For a full-queue audit, pause producers and other operators, inspect and+-- archive every row that requires retention, verify the active depth is zero, and+-- only then consider purge. A bounded read followed by purge can still delete+-- uninspected visible rows. 'purgeDlqForce' is the explicitly unconditional escape+-- hatch and permanently deletes hidden rows too. Alert on DLQ depth via+-- 'Keiro.PGMQ.Metrics.jobDlqMetrics'.+--+-- Redrive preserves the wrapper's original producer headers, including FIFO group+-- and trace metadata, but it sends a new main-queue row with a new id, a fresh read+-- count, and a new position at the back of its group. Redrive is at-least-once: a+-- crash after sending the original payload but before deleting the DLQ row leaves+-- the payload in both places. Handlers must therefore be idempotent. module Keiro.PGMQ.Dlq   ( DlqEntry (..),     readDlq,     redriveDlq,+    PurgeDlqResult (..),     purgeDlq,+    purgeDlqForce,     archiveDlq,+    archiveDlqEntries,     archiveDlqEntry,     archiveDlqEntryById,   )@@ -44,13 +67,17 @@ import "base" Data.Int (Int32, Int64) import "effectful-core" Effectful (Eff, IOE, (:>)) import "pgmq-effectful" Pgmq.Effectful-  ( Message (..),+  ( BatchMessageQuery (..),+    Message (..),     MessageBody (..),+    MessageHeaders (..),     MessageId (..),     MessageQuery (..),     Pgmq,+    QueueMetrics (..),     ReadMessage (..),     SendMessage (..),+    SendMessageWithHeaders (..),   ) import "pgmq-effectful" Pgmq.Effectful qualified as Pgmq import "text" Data.Text (Text)@@ -68,6 +95,8 @@     originalMessageId :: !(Maybe Int64),     originalEnqueuedAt :: !(Maybe UTCTime),     readCount :: !(Maybe Int64),+    -- | The producer headers preserved in the DLQ wrapper, when present.+    originalHeaders :: !(Maybe Value),     -- | Full DLQ wrapper for forensics.     rawBody :: !Value   }@@ -78,7 +107,8 @@     deadLetterReason :: !Text,     envelopeOriginalMessageId :: !(Maybe Int64),     envelopeOriginalEnqueuedAt :: !(Maybe UTCTime),-    envelopeReadCount :: !(Maybe Int64)+    envelopeReadCount :: !(Maybe Int64),+    envelopeOriginalHeaders :: !(Maybe Value)   }  parseDlqEnvelope :: Value -> Either Text DlqEnvelope@@ -93,13 +123,15 @@         envelopeOriginalMessageId <- obj .:? "original_message_id"         envelopeOriginalEnqueuedAt <- obj .:? "original_enqueued_at"         envelopeReadCount <- obj .:? "read_count"+        envelopeOriginalHeaders <- obj .:? "original_headers"         pure           DlqEnvelope             { originalMessage,               deadLetterReason,               envelopeOriginalMessageId,               envelopeOriginalEnqueuedAt,-              envelopeReadCount+              envelopeReadCount,+              envelopeOriginalHeaders             }      firstText = \case@@ -134,6 +166,7 @@               originalMessageId = Nothing,               originalEnqueuedAt = Nothing,               readCount = Nothing,+              originalHeaders = Nothing,               rawBody = body             }         Right envelope ->@@ -144,12 +177,14 @@               originalMessageId = envelope.envelopeOriginalMessageId,               originalEnqueuedAt = envelope.envelopeOriginalEnqueuedAt,               readCount = envelope.envelopeReadCount,+              originalHeaders = envelope.envelopeOriginalHeaders,               rawBody = body             } --- | Move up to @n@ DLQ rows back to the main queue. Redriven messages start a--- fresh PGMQ @read_ct@ on the main queue. Malformed DLQ wrappers are left in the--- DLQ for inspection.+-- | Move up to @n@ visible DLQ rows back to the main queue. Each read hides the+-- DLQ row for 30 seconds. Redriven messages preserve wrapper producer headers but+-- receive a new id, queue position, and fresh PGMQ @read_ct@. Malformed wrappers+-- are left in the DLQ for inspection. redriveDlq :: (Pgmq :> es, IOE :> es) => Job p -> Int -> Eff es Int redriveDlq job n   | n <= 0 = pure 0@@ -179,13 +214,22 @@         Left _err ->           pure count         Right envelope -> do-          _ <--            Pgmq.sendMessage-              SendMessage-                { queueName = job.jobQueue.physicalName,-                  messageBody = MessageBody envelope.originalMessage,-                  delay = Nothing-                }+          _ <- case envelope.envelopeOriginalHeaders of+            Just headers ->+              Pgmq.sendMessageWithHeaders+                SendMessageWithHeaders+                  { queueName = job.jobQueue.physicalName,+                    messageBody = MessageBody envelope.originalMessage,+                    messageHeaders = MessageHeaders headers,+                    delay = Nothing+                  }+            Nothing ->+              Pgmq.sendMessage+                SendMessage+                  { queueName = job.jobQueue.physicalName,+                    messageBody = MessageBody envelope.originalMessage,+                    delay = Nothing+                  }           void $             Pgmq.deleteMessage               MessageQuery@@ -194,12 +238,32 @@                 }           pure (count + 1) --- | Delete all rows currently in the DLQ.-purgeDlq :: (Pgmq :> es, IOE :> es) => Job p -> Eff es ()-purgeDlq job =-  void (Pgmq.deleteAllMessagesFromQueue job.jobQueue.dlqName)+-- | Result of a visibility-safe DLQ purge attempt.+data PurgeDlqResult+  = -- | The DLQ had no hidden rows at the metrics snapshot and was purged.+    PurgeDlqPurged !Int64+  | -- | Purge was refused because this many rows were hidden at the snapshot.+    PurgeDlqBlocked !Int64+  deriving stock (Eq, Show) --- | Archive (retain) up to @n@ DLQ rows: move each out of the active DLQ table+-- | Delete all rows currently in the DLQ only when none are hidden by a prior+-- read. The metrics check and deletion are separate operations, so callers that+-- need a full-queue audit must also quiesce concurrent readers and writers.+purgeDlq :: (Pgmq :> es, IOE :> es) => Job p -> Eff es PurgeDlqResult+purgeDlq job = do+  metrics <- Pgmq.queueMetrics job.jobQueue.dlqName+  let invisible = metrics.queueLength - metrics.queueVisibleLength+  if invisible > 0+    then pure (PurgeDlqBlocked invisible)+    else PurgeDlqPurged <$> Pgmq.deleteAllMessagesFromQueue job.jobQueue.dlqName++-- | Unconditionally delete every active DLQ row, including hidden rows.+purgeDlqForce :: (Pgmq :> es, IOE :> es) => Job p -> Eff es Int64+purgeDlqForce job =+  Pgmq.deleteAllMessagesFromQueue job.jobQueue.dlqName++-- | Archive (retain) up to @n@ visible DLQ rows, hiding each inspected row for+-- 30 seconds: move each out of the active DLQ table -- @pgmq.q_<dlq>@ into the archive table @pgmq.a_<dlq>@, preserving @enqueued_at@ / -- @read_ct@ and stamping @archived_at@. Returns the number archived. This is the -- audit-retention counterpart to the delete-only 'purgeDlq'. At-most-once per row@@ -231,6 +295,17 @@     archiveOne count message = do       moved <- archiveDlqEntry job message.messageId       pure (if moved then count + 1 else count)++-- | Archive the specified DLQ rows, including rows hidden by a prior read.+-- Returns only ids that were present and moved; returned order is unspecified.+archiveDlqEntries :: (Pgmq :> es, IOE :> es) => Job p -> [MessageId] -> Eff es [MessageId]+archiveDlqEntries _job [] = pure []+archiveDlqEntries job messageIds =+  Pgmq.batchArchiveMessages+    BatchMessageQuery+      { queueName = job.jobQueue.dlqName,+        messageIds+      }  -- | Archive one specific DLQ row by message id. 'True' if a row was moved. archiveDlqEntry :: (Pgmq :> es, IOE :> es) => Job p -> MessageId -> Eff es Bool

@@ -7,7 +7,7 @@ -- Integration Points and depended on by the two consumer migrations), and it -- keeps the producer signatures uniform with the processor/runner ones. We -- therefore silence the otherwise-correct redundant-constraint warning here.-{-# OPTIONS_GHC -Wno-redundant-constraints #-}+{-# OPTIONS_GHC -Wno-redundant-constraints -Wno-partial-fields #-}  -- | Layer 2 of @keiro-pgmq@: the typed-'Job' ergonomics built on top of -- 'Keiro.PGMQ.Runtime'. This is the payoff layer that absorbs the boilerplate two@@ -81,6 +81,7 @@     JobOrdering (..),     JobTuning (..),     JobTuningConfigError (..),+    JobConsumptionConfigError (..),     mkJobTuning,     defaultJobTuning,     withOrdering,@@ -104,6 +105,8 @@     -- * Queue lifecycle     QueueKind (..),     PartitionSpec (..),+    PartitionSpecConfigError (..),+    mkPartitionSpec,     QueueProvision (..),     standardProvision,     unloggedProvision,@@ -157,7 +160,7 @@     mergeTraceHeaders,   ) import "pgmq-effectful" Pgmq.Effectful qualified as Pgmq-import "pgmq-effectful" Pgmq.Effectful.Effect (readGrouped, readGroupedRoundRobin)+import "pgmq-effectful" Pgmq.Effectful.Effect (readGrouped, readGroupedHead, readGroupedRoundRobin) import "pgmq-hasql" Pgmq.Hasql.Statements.Types (ReadGrouped (..)) import "shibuya-core" Shibuya.App   ( AppConfig (..),@@ -225,6 +228,7 @@   ) import "text" Data.Text (Text) import "text" Data.Text qualified as Text+import "text" Data.Text.Read qualified as TextRead import "time" Data.Time (NominalDiffTime, nominalDiffTimeToSeconds)  -- | What a job handler decides. Never exposes shibuya/PGMQ wire types to the caller.@@ -297,17 +301,22 @@ -- selection order (@msg_id@ ascending), with NO per-key delivery-order guarantee -- under concurrent workers, retries, or visibility-timeout expiry. ----- 'FifoThroughput' and 'FifoRoundRobin' enable strict per-group ordering via PGMQ--- message groups (the reserved @x-pgmq-group@ header). Within one group, messages--- are delivered in strict send order; distinct groups proceed in parallel.+-- 'FifoHeads' enables strict per-group ordering via PGMQ message groups (the+-- reserved @x-pgmq-group@ header). It returns at most one absolute head per+-- group, so a failed or delayed head blocks its successors while distinct groups+-- stay independently eligible. Current Keiro handlers remain serial.+-- -- 'FifoThroughput' fills a batch from the oldest eligible group first (SQS-style, -- @read_grouped@); 'FifoRoundRobin' interleaves fairly across groups--- (@read_grouped_rr@). Delivery is still at-least-once and there is no--- deduplication, so handlers must be idempotent.+-- (@read_grouped_rr@). Those legacy strategies may safely use only a batch size+-- of one; larger batches can lease a successor before its predecessor settles.+-- Delivery is still at-least-once and there is no deduplication, so handlers+-- must be idempotent. data JobOrdering   = Unordered   | FifoThroughput   | FifoRoundRobin+  | FifoHeads   deriving stock (Eq, Show)  -- | How a consumer reads the queue.@@ -356,6 +365,37 @@ validPolling (LongPoll maxPollSeconds pollIntervalMs) =   maxPollSeconds > 0 && pollIntervalMs > 0 +-- | Invalid or contradictory consumer configuration. These checks run before+-- an adapter is constructed or a direct PGMQ read is issued, including when raw+-- 'JobTuning' and 'Job' constructors are used.+data JobConsumptionConfigError+  = InvalidJobTuning !JobTuningConfigError+  | JobOrderingMismatch+      { jobOrderingDeclared :: !JobOrdering,+        tuningOrderingGiven :: !JobOrdering+      }+  | UnsafeLegacyFifoBatch+      { unsafeOrdering :: !JobOrdering,+        unsafeBatchSize :: !Int32+      }+  deriving stock (Eq, Show)+  deriving anyclass (Exception)++validateJobConsumptionConfig :: Job p -> JobTuning -> Either JobConsumptionConfigError ()+validateJobConsumptionConfig job tuning = do+  case mkJobTuning tuning.visibilityTimeout tuning.batchSize tuning.polling of+    Left err -> Left (InvalidJobTuning err)+    Right _ -> Right ()+  if job.jobOrdering == tuning.ordering+    then Right ()+    else Left JobOrderingMismatch {jobOrderingDeclared = job.jobOrdering, tuningOrderingGiven = tuning.ordering}+  case tuning.ordering of+    FifoThroughput | tuning.batchSize > 1 -> unsafe+    FifoRoundRobin | tuning.batchSize > 1 -> unsafe+    _ -> Right ()+  where+    unsafe = Left UnsafeLegacyFifoBatch {unsafeOrdering = tuning.ordering, unsafeBatchSize = tuning.batchSize}+ toPollingConfig :: JobPolling -> PollingConfig toPollingConfig (PollEvery interval) = StandardPolling interval toPollingConfig (LongPoll maxPollSeconds pollIntervalMs) = LongPolling maxPollSeconds pollIntervalMs@@ -365,6 +405,7 @@ toFifoConfig Unordered = Nothing toFifoConfig FifoThroughput = Just (FifoConfig ThroughputOptimized) toFifoConfig FifoRoundRobin = Just (FifoConfig RoundRobin)+toFifoConfig FifoHeads = Just (FifoConfig HeadPerGroup)  retryDelaySeconds :: RetryDelay -> NominalDiffTime retryDelaySeconds (RetryDelay seconds) = seconds@@ -380,14 +421,16 @@       clamped = max minSec (min maxSec seconds)    in ceiling clamped --- | A declarative job: a queue, a payload codec, and a retry policy, named for--- telemetry. Construct one and pair it with a handler of type+-- | A declarative job: a queue, a payload codec, an ordering contract, and a+-- retry policy, named for telemetry. Construct one and pair it with a handler of type -- @p -> Eff es 'JobOutcome'@. data Job p = Job   { -- | Used as the shibuya 'ProcessorId' and telemetry label.     jobName :: !Text,     jobQueue :: !QueueRef,     jobCodec :: !(JobCodec p),+    -- | Required consumption ordering. Explicit tuning must match this value.+    jobOrdering :: !JobOrdering,     jobPolicy :: !RetryPolicy   } @@ -529,9 +572,9 @@ -- | Enqueue a payload into the FIFO group named by @groupKey@. The group key is -- written under the reserved @x-pgmq-group@ JSONB header, which PGMQ's grouped -- reads and the shibuya adapter use to order deliveries per group. Consume with an--- ordered 'JobTuning' (see 'withOrdering') to honor the order; within one group,--- messages are handled in strict send order while distinct groups proceed in--- parallel.+-- ordered 'JobTuning' (see 'withOrdering') to honor the order. With 'FifoHeads',+-- messages are handled in strict send order within one group while distinct+-- groups can be claimed together; current Keiro handlers remain serial. enqueueToGroup ::   (Pgmq :> es, IOE :> es) => Job p -> Text -> p -> Eff es MessageId enqueueToGroup job groupKey p =@@ -554,25 +597,92 @@ --     table is truncated to empty on a database crash. For transient, regenerable --     work. --   * 'PartitionedKind' — storage split across child tables by time or message-id---     range, managed by the PostgreSQL extension @pg_partman@. Requires a---     @pg_partman@-enabled server (see 'partitionedProvision').+--     range, managed by the PostgreSQL extension @pg_partman@. This mode is+--     experimental and requires a @pg_partman@-enabled server. Retention drops+--     whole old partitions, including unprocessed active rows (see+--     'partitionedProvision'). data QueueKind   = StandardKind   | UnloggedKind   | PartitionedKind !PartitionSpec   deriving stock (Eq, Show) --- | Partition interval + retention interval for a partitioned queue. Both are--- PostgreSQL/@pg_partman@ duration or integer strings — e.g. @"daily"@ or--- @"10000"@ for the interval, @"7 days"@ or @"100000"@ for the retention.+-- | Partition interval and retention policy for a partitioned queue. PGMQ passes+-- both values to PostgreSQL/@pg_partman@: time values are opaque strings such as+-- @"daily"@ and @"7 days"@, while positive PostgreSQL @INTEGER@ text selects+-- message-id-range partitioning.+--+-- Retention is not per-message expiry. Maintenance permanently drops whole old+-- partitions from both the active queue and its archive without checking whether+-- an active message was processed. Size retention above the worst expected+-- consumer outage and backlog age, with an operator-chosen safety margin.+--+-- The raw constructor is intentionally available for server-specific values but+-- is unvalidated. Prefer 'mkPartitionSpec' for the bounded checks Keiro can make+-- without duplicating PostgreSQL's interval parser. data PartitionSpec = PartitionSpec   { partitionInterval :: !Text,     retentionInterval :: !Text   }   deriving stock (Eq, Show) +-- | Why 'mkPartitionSpec' rejected a partition configuration.+data PartitionSpecConfigError+  = EmptyPartitionInterval+  | EmptyRetentionInterval+  | NonPositivePartitionInterval !Integer+  | NonPositiveRetentionInterval !Integer+  | PartitionIntervalOutsidePostgresInteger !Integer+  | MixedPartitionUnits !Text !Text+  | RetentionBelowPartitionInterval !Integer !Integer+  deriving stock (Eq, Show)++-- | Construct a partition specification after applying Keiro's bounded+-- preflight checks. Inputs are trimmed. Fully numeric values must use matching+-- message-id units; the partition span must fit PostgreSQL's positive signed+-- 32-bit @INTEGER@ classifier, and numeric retention must cover at least one+-- configured span. Nonnumeric pairs remain opaque for PostgreSQL/@pg_partman@ to+-- validate.+--+-- These checks reject obvious mistakes; they do not prove a safe retention+-- horizon. Operators must still account for worst-case outage and backlog age.+mkPartitionSpec :: Text -> Text -> Either PartitionSpecConfigError PartitionSpec+mkPartitionSpec rawPartition rawRetention+  | Text.null partition = Left EmptyPartitionInterval+  | Text.null retention = Left EmptyRetentionInterval+  | otherwise =+      case (completeInteger partition, completeInteger retention) of+        (Just partitionValue, Just retentionValue)+          | partitionValue <= 0 -> Left (NonPositivePartitionInterval partitionValue)+          | partitionValue > toInteger (maxBound :: Int32) ->+              Left (PartitionIntervalOutsidePostgresInteger partitionValue)+          | retentionValue <= 0 -> Left (NonPositiveRetentionInterval retentionValue)+          | retentionValue < partitionValue ->+              Left (RetentionBelowPartitionInterval partitionValue retentionValue)+          | otherwise -> Right spec+        (Just partitionValue, Nothing)+          | partitionValue <= 0 -> Left (NonPositivePartitionInterval partitionValue)+          | partitionValue > toInteger (maxBound :: Int32) ->+              Left (PartitionIntervalOutsidePostgresInteger partitionValue)+          | otherwise -> Left (MixedPartitionUnits partition retention)+        (Nothing, Just retentionValue)+          | retentionValue <= 0 -> Left (NonPositiveRetentionInterval retentionValue)+          | otherwise -> Left (MixedPartitionUnits partition retention)+        (Nothing, Nothing) -> Right spec+  where+    partition = Text.strip rawPartition+    retention = Text.strip rawRetention+    spec = PartitionSpec {partitionInterval = partition, retentionInterval = retention}++    completeInteger value =+      case TextRead.signed TextRead.decimal value of+        Right (parsed, remainder) | Text.null remainder -> Just parsed+        _ -> Nothing+ -- | The provisioning choice for a job's /main/ queue: which storage shape, and--- whether to create the FIFO GIN index. The DLQ (when the policy enables one) is+-- whether to create PGMQ's conventional FIFO GIN index on @headers@. Presence is+-- reported by the upstream index name; it does not prove that PostgreSQL uses the+-- index for grouped-read expressions. The DLQ (when the policy enables one) is -- always a plain standard queue with no FIFO index. data QueueProvision = QueueProvision   { provisionKind :: !QueueKind,@@ -588,12 +698,17 @@ unloggedProvision :: QueueProvision unloggedProvision = QueueProvision {provisionKind = UnloggedKind, provisionFifoIndex = False} --- | A partitioned main queue (no FIFO index) with the given interval/retention.+-- | An experimental partitioned main queue (no FIFO index) with the given+-- interval and retention. Maintenance drops whole old active and archive+-- partitions, so unprocessed work can be lost when retention is shorter than an+-- outage or backlog. Prefer a value built by 'mkPartitionSpec'. partitionedProvision :: PartitionSpec -> QueueProvision partitionedProvision spec =   QueueProvision {provisionKind = PartitionedKind spec, provisionFifoIndex = False} --- | Turn on FIFO-index creation for a provisioning choice.+-- | Request PGMQ's conventional FIFO GIN index for a provisioning choice. This+-- preserves upstream provisioning and presence-reporting behavior; it is not a+-- claim that the index accelerates every grouped-read query. withFifoIndexProvision :: QueueProvision -> QueueProvision withFifoIndexProvision provision = provision {provisionFifoIndex = True} @@ -615,7 +730,8 @@             job.jobQueue.physicalName             Config.PartitionConfig               { Config.partitionInterval = spec.partitionInterval,-                Config.retentionInterval = spec.retentionInterval+                Config.retentionInterval = spec.retentionInterval,+                Config.premake = Nothing               }     mainConfig       | provision.provisionFifoIndex = Config.withFifoIndex mainBase@@ -639,13 +755,18 @@ ensureJobQueue :: (Pgmq :> es) => Job p -> Eff es () ensureJobQueue = ensureJobQueueWith standardProvision --- | Create the FIFO GIN index on the job's /main/ queue's @headers@ column —--- the index PGMQ's grouped/ordered reads (@read_grouped@/@read_grouped_rr@) match--- against. Idempotent: the index step is always re-applied and the underlying SQL--- is @CREATE INDEX IF NOT EXISTS@, so a second call is a harmless no-op. Routing+-- | Request PGMQ's conventional FIFO GIN index on the job's /main/ queue's+-- @headers@ column. PGMQ reports presence by the conventional index name; neither+-- presence nor successful creation proves that PostgreSQL uses it for the group+-- expression in @read_grouped@, @read_grouped_rr@, or @read_grouped_head@. No+-- supplemental expression index is currently recommended without reproducible+-- workload measurements.+--+-- Idempotent: the index step is always re-applied and the underlying SQL is+-- @CREATE INDEX IF NOT EXISTS@, so a second call is a harmless no-op. Routing -- through @pgmq-config@'s reconciler (which lists existing queues first) means -- calling this on an already-provisioned queue does not recreate the queue. This--- is the artifact the FIFO ordered-delivery plan+-- is the conventional provisioning artifact the FIFO ordered-delivery plan -- (@docs/plans/77-add-fifo-ordered-delivery-via-message-groups-to-keiro-pgmq.md@) -- consumes for ordered jobs. ensureFifoIndex :: (Pgmq :> es) => Job p -> Eff es ()@@ -654,9 +775,10 @@     [Config.withFifoIndex (Config.standardQueue job.jobQueue.physicalName)]  -- | Provision an ordered job's queue: create the main queue (and the DLQ when--- the policy uses one) plus the FIFO GIN index that grouped reads need. Composes--- 'ensureJobQueue' and 'ensureFifoIndex'; both are idempotent, so this is safe to--- call at every startup.+-- the policy uses one) plus PGMQ's conventional FIFO GIN index. This preserves+-- upstream's ordered-queue setup but does not claim grouped-read acceleration.+-- Composes 'ensureJobQueue' and 'ensureFifoIndex'; both are idempotent, so this is+-- safe to call at every startup. ensureOrderedJobQueue :: (Pgmq :> es) => Job p -> Eff es () ensureOrderedJobQueue job = do   ensureJobQueue job@@ -731,6 +853,7 @@   (JobContext es -> p -> Eff es JobOutcome) ->   Eff es (ProcessorId, QueueProcessor es) jobProcessorWithContext tuning job handle = do+  either (liftIO . throwIO) pure (validateJobConsumptionConfig job tuning)   env <- ask   adapter <-     pgmqAdapter env (adapterConfigFor tuning job)@@ -752,7 +875,7 @@   (p -> Eff es JobOutcome) ->   Eff es (ProcessorId, QueueProcessor es) jobProcessor job handle =-  jobProcessorWithContext defaultJobTuning job (\_context p -> handle p)+  jobProcessorWithContext (withOrdering job.jobOrdering defaultJobTuning) job (\_context p -> handle p)  -- | Continuous, multi-processor run (the @rei@ cadence): run a supervised app -- over several processors built with 'jobProcessor'. Returns the app handle; the@@ -865,9 +988,9 @@   Job p ->   (JobContext es -> p -> Eff es JobOutcome) ->   Eff es Int-runJobOnceWithContext tuning n job handle-  | n <= 0 = pure 0-  | otherwise = drain 0+runJobOnceWithContext tuning n job handle = do+  either (liftIO . throwIO) pure (validateJobConsumptionConfig job tuning)+  if n <= 0 then pure 0 else drain 0   where     drain handled       | handled >= n = pure handled@@ -896,6 +1019,13 @@                     visibilityTimeout = tuning.visibilityTimeout,                     qty = qty                   }+            FifoHeads ->+              readGroupedHead+                ReadGrouped+                  { queueName = job.jobQueue.physicalName,+                    visibilityTimeout = tuning.visibilityTimeout,+                    qty = qty+                  }           if null messages             then pure handled             else do@@ -1042,7 +1172,7 @@ runJobOnce n job handle =   void $     runJobOnceWithContext-      defaultJobTuning+      (withOrdering job.jobOrdering defaultJobTuning)       n       job       (\_context p -> handle p)

@@ -7,8 +7,9 @@ -- queue and its DEAD-LETTER queue without the caller deriving any physical name. -- -- Use 'jobDlqMetrics' (its 'queueLength') for the depth alerting that--- 'Keiro.PGMQ.Dlq' recommends; pair it with 'Keiro.PGMQ.Dlq.archiveDlq' /--- 'Keiro.PGMQ.Dlq.purgeDlq' for retention.+-- 'Keiro.PGMQ.Dlq' recommends; after inspection, pair it with+-- 'Keiro.PGMQ.Dlq.archiveDlqEntries' and inspect the typed+-- 'Keiro.PGMQ.Dlq.PurgeDlqResult' returned by 'Keiro.PGMQ.Dlq.purgeDlq'. module Keiro.PGMQ.Metrics   ( QueueMetrics (..),     jobQueueMetrics,

@@ -4,6 +4,7 @@ {-# LANGUAGE DerivingStrategies #-} {-# LANGUAGE LambdaCase #-} {-# LANGUAGE OverloadedRecordDot #-}+{-# LANGUAGE TypeApplications #-}  -- | End-to-end integration test for @keiro-pgmq@. --@@ -17,7 +18,7 @@ module Main (main) where  import Control.Concurrent (threadDelay)-import Control.Exception (bracket, throwIO)+import Control.Exception (bracket, throwIO, try) import Data.Aeson (FromJSON, ToJSON, Value (..), object, parseJSON, toJSON, (.=)) import Data.Aeson.Key qualified as Key import Data.Aeson.KeyMap qualified as KeyMap@@ -57,7 +58,7 @@ import OpenTelemetry.Trace.Id.Generator.Default (defaultIdGenerator) import OpenTelemetry.Util (appendOnlyBoundedCollectionValues) import Pgmq.Config.Types qualified as Config-import Pgmq.Effectful (Message (..), MessageBody (..), Pgmq, QueueMetrics (..), ReadMessage (..), SendMessage (..))+import Pgmq.Effectful (Message (..), MessageBody (..), Pgmq, ReadMessage (..), SendMessage (..)) import Pgmq.Effectful qualified as Pgmq import Pgmq.Migration qualified as Migration import Pgmq.Types (QueueName, parseQueueName, queueNameToText)@@ -130,6 +131,7 @@     { jobName = name,       jobQueue = queueRef name,       jobCodec = aesonJobCodec,+      jobOrdering = Unordered,       jobPolicy = defaultRetryPolicy     } @@ -376,6 +378,33 @@     mkJobTuning 30 1 (PollEvery 1)       `shouldBe` Right defaultJobTuning +  it "validates partition specifications without parsing time intervals" $ \_connStr -> do+    mkPartitionSpec " daily " " 7 days "+      `shouldBe` Right (PartitionSpec "daily" "7 days")+    mkPartitionSpec " 10000 " " 100000 "+      `shouldBe` Right (PartitionSpec "10000" "100000")+    mkPartitionSpec "" "7 days"+      `shouldBe` Left EmptyPartitionInterval+    mkPartitionSpec "daily" "   "+      `shouldBe` Left EmptyRetentionInterval+    mkPartitionSpec "0" "100"+      `shouldBe` Left (NonPositivePartitionInterval 0)+    mkPartitionSpec "-1" "100"+      `shouldBe` Left (NonPositivePartitionInterval (-1))+    mkPartitionSpec "100" "0"+      `shouldBe` Left (NonPositiveRetentionInterval 0)+    mkPartitionSpec "100" "-1"+      `shouldBe` Left (NonPositiveRetentionInterval (-1))+    mkPartitionSpec "2147483648" "2147483648"+      `shouldBe` Left (PartitionIntervalOutsidePostgresInteger 2147483648)+    mkPartitionSpec "100" "7 days"+      `shouldBe` Left (MixedPartitionUnits "100" "7 days")+    mkPartitionSpec "daily" "100"+      `shouldBe` Left (MixedPartitionUnits "daily" "100")+    mkPartitionSpec "100" "99"+      `shouldBe` Left (RetentionBelowPartitionInterval 100 99)+    PartitionSpec "" "" `shouldBe` PartitionSpec "" ""+   it "derives distinct physical names for long logical queue names" $ \_connStr -> do     let commonPrefix = Text.replicate 43 "a"         first = queueRef (commonPrefix <> "x")@@ -688,10 +717,11 @@   it "readDlq decodes the original dead-lettered payload" $ \connStr -> do     let job = mkJob "keiro_pgmq_test.dlq_read"         payload = Ping "poison" 3+        headers = object ["tenant" .= ("acme" :: Text)]     entries <-       runDb connStr $ do         ensureJobQueue job-        _ <- enqueue job payload+        _ <- enqueueWithHeaders job (MessageHeaders headers) payload         runJobOnce 1 job (\_ -> pure (Dead "bad"))         readDlq job 1     case entries of@@ -700,6 +730,7 @@         entry.originalPayload `shouldBe` Right payload         entry.originalMessageId `shouldSatisfy` (/= Nothing)         entry.readCount `shouldBe` Just 1+        entry.originalHeaders `shouldBe` Just headers       _ -> expectationFailure ("expected one DLQ entry, got " <> show (length entries))    it "redriveDlq moves dead-lettered payloads back to the main queue" $ \connStr -> do@@ -720,16 +751,127 @@     finalMainLen <- runDb connStr (queueLen job.jobQueue.physicalName)     finalMainLen `shouldBe` 0 -  it "purgeDlq empties the DLQ" $ \connStr -> do+  it "redriveDlq preserves the wrapper's original headers" $ \connStr -> do+    let job = mkJob "keiro_pgmq_test.dlq_redrive_headers"+        originalHeaders =+          object+            [ "x-pgmq-group" .= ("orders" :: Text),+              "traceparent" .= ("00-4bf92f3577b34da6a3ce929d0e0e4736-00f067aa0ba902b7-01" :: Text),+              "tenant" .= ("acme" :: Text)+            ]+    messages <-+      runDb connStr $ do+        ensureJobQueue job+        _ <- enqueueWithHeaders job (MessageHeaders originalHeaders) (Ping "redrive-headers" 1)+        runJobOnce 1 job (\_ -> pure (Dead "bad"))+        redriven <- redriveDlq job 1+        liftIO (redriven `shouldBe` 1)+        readMessages job.jobQueue.physicalName 1+    case messages of+      [message] -> message.headers `shouldBe` Just originalHeaders+      _ -> expectationFailure ("expected one redriven message, got " <> show (length messages))++  it "redriveDlq uses wrapper headers instead of the DLQ row's headers" $ \connStr -> do+    let job = mkJob "keiro_pgmq_test.dlq_redrive_wrapper_headers"+        originalHeaders = object ["tenant" .= ("original" :: Text)]+        dlqRowHeaders = object ["tenant" .= ("dlq-row" :: Text)]+        wrapper =+          object+            [ "original_message" .= Ping "wrapper-headers" 1,+              "dead_letter_reason" .= ("poison_pill: bad" :: Text),+              "original_headers" .= originalHeaders+            ]+    messages <-+      runDb connStr $ do+        ensureJobQueue job+        _ <-+          Pgmq.sendMessageWithHeaders+            Pgmq.SendMessageWithHeaders+              { queueName = job.jobQueue.dlqName,+                messageBody = MessageBody wrapper,+                messageHeaders = MessageHeaders dlqRowHeaders,+                delay = Nothing+              }+        _ <- redriveDlq job 1+        readMessages job.jobQueue.physicalName 1+    case messages of+      [message] -> message.headers `shouldBe` Just originalHeaders+      _ -> expectationFailure ("expected one redriven message, got " <> show (length messages))++  it "redriveDlq keeps missing and null wrapper headers headerless" $ \connStr -> do+    let rowHeaders = MessageHeaders (object ["tenant" .= ("dlq-row" :: Text)])+        redriveLegacy queue wrapper = do+          let job = mkJob queue+          runDb connStr $ do+            ensureJobQueue job+            _ <-+              Pgmq.sendMessageWithHeaders+                Pgmq.SendMessageWithHeaders+                  { queueName = job.jobQueue.dlqName,+                    messageBody = MessageBody wrapper,+                    messageHeaders = rowHeaders,+                    delay = Nothing+                  }+            _ <- redriveDlq job 1+            readMessages job.jobQueue.physicalName 1+        missingHeaders =+          object+            [ "original_message" .= Ping "missing-headers" 1,+              "dead_letter_reason" .= ("poison_pill: bad" :: Text)+            ]+        nullHeaders =+          object+            [ "original_message" .= Ping "null-headers" 1,+              "dead_letter_reason" .= ("poison_pill: bad" :: Text),+              "original_headers" .= Null+            ]+    missingMessages <- redriveLegacy "keiro_pgmq_test.dlq_redrive_missing_headers" missingHeaders+    nullMessages <- redriveLegacy "keiro_pgmq_test.dlq_redrive_null_headers" nullHeaders+    map (.headers) missingMessages `shouldBe` [Nothing]+    map (.headers) nullMessages `shouldBe` [Nothing]++  it "purgeDlq empties a fully visible DLQ and reports the deleted count" $ \connStr -> do     let job = mkJob "keiro_pgmq_test.dlq_purge"-    runDb connStr $ do+    result <- runDb connStr $ do       ensureJobQueue job       _ <- enqueue job (Ping "purge" 1)       runJobOnce 1 job (\_ -> pure (Dead "bad"))       purgeDlq job+    result `shouldBe` PurgeDlqPurged 1     dlqLen <- runDb connStr (queueLen job.jobQueue.dlqName)     dlqLen `shouldBe` 0 +  it "purgeDlq refuses when inspection has hidden a row and deletes nothing" $ \connStr -> do+    let job = mkJob "keiro_pgmq_test.dlq_purge_blocked"+    (result, remaining) <-+      runDb connStr $ do+        ensureJobQueue job+        _ <- enqueue job (Ping "hidden" 1)+        _ <- enqueue job (Ping "visible" 2)+        runJobOnce 2 job (\_ -> pure (Dead "bad"))+        inspected <- readDlq job 1+        liftIO (length inspected `shouldBe` 1)+        result <- purgeDlq job+        remaining <- queueLen job.jobQueue.dlqName+        pure (result, remaining)+    result `shouldBe` PurgeDlqBlocked 1+    remaining `shouldBe` 2++  it "purgeDlqForce deletes hidden rows" $ \connStr -> do+    let job = mkJob "keiro_pgmq_test.dlq_purge_force"+    (purged, remaining) <-+      runDb connStr $ do+        ensureJobQueue job+        _ <- enqueue job (Ping "hidden" 1)+        runJobOnce 1 job (\_ -> pure (Dead "bad"))+        inspected <- readDlq job 1+        liftIO (length inspected `shouldBe` 1)+        purged <- purgeDlqForce job+        remaining <- queueLen job.jobQueue.dlqName+        pure (purged, remaining)+    purged `shouldBe` 1+    remaining `shouldBe` 0+   it "readDlq preserves malformed DLQ wrappers as malformed entries" $ \connStr -> do     let job = mkJob "keiro_pgmq_test.dlq_malformed"     entries <-@@ -749,6 +891,7 @@         entry.originalPayload `shouldSatisfy` \case           Left (JobPayloadMalformed _) -> True           _ -> False+        entry.originalHeaders `shouldBe` Nothing       _ -> expectationFailure ("expected one malformed DLQ entry, got " <> show (length entries))    it "undecodable payload routes to the DLQ" $ \connStr -> do@@ -1033,7 +1176,7 @@   it "a FIFO delivery carries shibuya.partition on its process span" $ \connStr -> do     (provider, spansRef) <- setupCapturingProvider     let tracer = OTel.makeTracer provider "keiro-pgmq-test" OTel.tracerOptions-        job = mkJob "keiro_pgmq_test.span_partition"+        job = (mkJob "keiro_pgmq_test.span_partition") {jobOrdering = FifoThroughput}     drained <-       runDbTraced connStr tracer $ do         ensureOrderedJobQueue job@@ -1066,13 +1209,14 @@   -- EP-2 M2: partitioned config shape (pure) + pending live test.   it "ensureJobQueueWith partitioned builds a partitioned QueueConfig" $ \_connStr -> do     let job = mkJob "keiro_pgmq_test.partitioned"-        spec = PartitionSpec {partitionInterval = "daily", retentionInterval = "7 days"}-    case queueProvisionConfigs (partitionedProvision spec) job of+        partitionSpec = either (error . show) id (mkPartitionSpec "daily" "7 days")+    case queueProvisionConfigs (partitionedProvision partitionSpec) job of       (mainCfg : _) ->         case mainCfg.queueType of           Config.PartitionedQueue pc -> do             pc.partitionInterval `shouldBe` "daily"             pc.retentionInterval `shouldBe` "7 days"+            pc.premake `shouldBe` Nothing             mainCfg.queueName `shouldBe` job.jobQueue.physicalName           other ->             expectationFailure@@ -1110,7 +1254,7 @@       _ -> expectationFailure ("expected one message, got " <> show (length msgs))    it "ensureOrderedJobQueue is idempotent and the queue accepts grouped work" $ \connStr -> do-    let job = mkJob "keiro_pgmq_test.ordered_setup"+    let job = (mkJob "keiro_pgmq_test.ordered_setup") {jobOrdering = FifoThroughput}     len <-       runDb connStr $ do         ensureOrderedJobQueue job@@ -1122,10 +1266,263 @@         queueLen job.jobQueue.physicalName     len `shouldBe` 0 +  it "rejects invalid, mismatched, and unsafe raw consumption tuning before a read" $ \connStr -> do+    let unorderedJob = mkJob "keiro_pgmq_test.invalid_consumption"+        headsJob = unorderedJob {jobOrdering = FifoHeads}+        throughputJob = unorderedJob {jobOrdering = FifoThroughput}+        invalidTuning =+          JobTuning+            { visibilityTimeout = 30,+              batchSize = 0,+              polling = PollEvery 1,+              ordering = Unordered+            }+        mismatchedTuning = defaultJobTuning+        unsafeTuning =+          withOrdering FifoThroughput $+            either (error . show) id $+              mkJobTuning 30 8 (PollEvery 1)+        run tuning job =+          runDb connStr $+            runJobOnceWithContext tuning 0 job \_ctx _payload -> pure Done+    invalid <- try @JobConsumptionConfigError (run invalidTuning unorderedJob)+    mismatch <- try @JobConsumptionConfigError (run mismatchedTuning headsJob)+    unsafeBatch <- try @JobConsumptionConfigError (run unsafeTuning throughputJob)+    invalid `shouldBe` Left (InvalidJobTuning (NonPositiveBatchSize 0))+    mismatch+      `shouldBe` Left+        JobOrderingMismatch+          { jobOrderingDeclared = FifoHeads,+            tuningOrderingGiven = Unordered+          }+    unsafeBatch+      `shouldBe` Left+        UnsafeLegacyFifoBatch+          { unsafeOrdering = FifoThroughput,+            unsafeBatchSize = 8+          }++  it "accepts large batches for unordered and grouped-head jobs" $ \connStr -> do+    let unorderedJob = mkJob "keiro_pgmq_test.valid_unordered_batch"+        headsJob = (mkJob "keiro_pgmq_test.valid_heads_batch") {jobOrdering = FifoHeads}+        unorderedTuning = either (error . show) id $ mkJobTuning 30 50 (PollEvery 1)+        headsTuning = withOrdering FifoHeads unorderedTuning+        run tuning job =+          runDb connStr $+            runJobOnceWithContext tuning 0 job \_ctx _payload -> pure Done+    run unorderedTuning unorderedJob `shouldReturn` 0+    run headsTuning headsJob `shouldReturn` 0+    headsTuning.batchSize `shouldBe` 50++  it "rejects mismatched ordering before emitting a receive span" $ \connStr -> do+    (provider, spansRef) <- setupCapturingProvider+    let tracer = OTel.makeTracer provider "keiro-pgmq-test" OTel.tracerOptions+        job = (mkJob "keiro_pgmq_test.mismatch_no_receive") {jobOrdering = FifoHeads}+        receiveName = "receive " <> queueNameToText job.jobQueue.physicalName+    result <-+      try @JobConsumptionConfigError $+        runDbTraced connStr tracer $+          runJobOnceWithContext defaultJobTuning 1 job \_ctx _payload -> pure Done+    case result of+      Left JobOrderingMismatch {} -> pure ()+      Left err -> expectationFailure ("unexpected mismatch error: " <> show err)+      Right _ -> expectationFailure "expected mismatched ordering to fail"+    spans <- capturedSpans provider spansRef+    map csName spans `shouldNotSatisfy` elem receiveName++  it "worker construction rejects invalid raw tuning and mismatched ordering before adapter use" $ \connStr -> do+    let job = (mkJob "keiro_pgmq_test.worker_mismatch") {jobOrdering = FifoHeads}+        invalidTuning =+          JobTuning+            { visibilityTimeout = 0,+              batchSize = 1,+              polling = PollEvery 1,+              ordering = FifoHeads+            }+        build tuning =+          runDb connStr $+            jobProcessorWithContext tuning job \_ctx _payload -> pure Done+    invalid <- try @JobConsumptionConfigError (build invalidTuning)+    mismatch <-+      try @JobConsumptionConfigError $+        build defaultJobTuning+    case invalid of+      Left (InvalidJobTuning (NonPositiveVisibilityTimeout 0)) -> pure ()+      Left err -> expectationFailure ("unexpected invalid-tuning error: " <> show err)+      Right _ -> expectationFailure "expected worker construction to reject raw invalid tuning"+    case mismatch of+      Left JobOrderingMismatch {jobOrderingDeclared = FifoHeads, tuningOrderingGiven = Unordered} -> pure ()+      Left err -> expectationFailure ("unexpected validation error: " <> show err)+      Right _ -> expectationFailure "expected worker construction to reject mismatched ordering"++  it "FifoHeads blocks a failed group head while other groups continue" $ \connStr -> do+    firstA <- newIORef True+    observed <- newIORef ([] :: [Text])+    let job = (mkJob "keiro_pgmq_test.fifo_heads_failure") {jobOrdering = FifoHeads}+        tuning =+          withOrdering FifoHeads $+            either (error . show) id $+              mkJobTuning 1 50 (PollEvery 0.1)+        handler _ctx payload = do+          liftIO $ modifyIORef' observed (<> [payload.message])+          shouldFail <- liftIO $ do+            firstPending <- readIORef firstA+            if payload.message == "a1" && firstPending+              then writeIORef firstA False >> pure True+              else pure False+          if shouldFail+            then liftIO (throwIO (userError "fail a1 once"))+            else pure Done+    (firstDrain, independentDrain) <-+      runDb connStr $ do+        ensureOrderedJobQueue job+        _ <- enqueueToGroup job "a" (Ping "a1" 1)+        _ <- enqueueToGroup job "a" (Ping "a2" 2)+        _ <- enqueueToGroup job "b" (Ping "b1" 1)+        _ <- enqueueToGroup job "b" (Ping "b2" 2)+        firstDrain <- runJobOnceWithContext tuning 2 job handler+        independentDrain <- runJobOnceWithContext tuning 2 job handler+        pure (firstDrain, independentDrain)+    firstDrain `shouldBe` 2+    independentDrain `shouldBe` 0+    readIORef observed `shouldReturn` ["a1", "b1", "b2"]+    threadDelay 1_100_000+    recovered <- runDb connStr $ runJobOnceWithContext tuning 2 job handler+    recovered `shouldBe` 2+    readIORef observed `shouldReturn` ["a1", "b1", "b2", "a1", "a2"]++  it "FifoHeads blocks a group successor when its head returns Retry" $ \connStr -> do+    observed <- newIORef ([] :: [Text])+    let job = (mkJob "keiro_pgmq_test.fifo_heads_retry") {jobOrdering = FifoHeads}+        tuning =+          withOrdering FifoHeads $+            either (error . show) id $+              mkJobTuning 30 50 (PollEvery 0.1)+        handler _ctx payload = do+          liftIO $ modifyIORef' observed (<> [payload.message])+          pure $+            if payload.message == "a1"+              then Retry (RetryDelay 30)+              else Done+    (firstDrain, blockedDrain) <-+      runDb connStr $ do+        ensureOrderedJobQueue job+        _ <- enqueueToGroup job "a" (Ping "a1" 1)+        _ <- enqueueToGroup job "a" (Ping "a2" 2)+        _ <- enqueueToGroup job "b" (Ping "b1" 1)+        firstDrain <- runJobOnceWithContext tuning 2 job handler+        blockedDrain <- runJobOnceWithContext tuning 1 job handler+        pure (firstDrain, blockedDrain)+    firstDrain `shouldBe` 2+    blockedDrain `shouldBe` 0+    readIORef observed `shouldReturn` ["a1", "b1"]++  it "FifoHeads drains sixteen independent heads with one receive" $ \connStr -> do+    (provider, spansRef) <- setupCapturingProvider+    let tracer = OTel.makeTracer provider "keiro-pgmq-test" OTel.tracerOptions+        job = (mkJob "keiro_pgmq_test.fifo_heads_batch") {jobOrdering = FifoHeads}+        tuning =+          withOrdering FifoHeads $+            either (error . show) id $+              mkJobTuning 30 16 (PollEvery 0.1)+        messages = [("g" <> Text.pack (show i), Ping ("m" <> Text.pack (show i)) i) | i <- [1 .. 16]]+        receiveName = "receive " <> queueNameToText job.jobQueue.physicalName+    drained <-+      runDbTraced connStr tracer $ do+        ensureOrderedJobQueue job+        traverse_ (\(groupName, payload) -> enqueueToGroup job groupName payload) messages+        runJobOnceWithContext tuning 16 job \_ctx _payload -> pure Done+    drained `shouldBe` 16+    spans <- capturedSpans provider spansRef+    length (filter ((== receiveName) . csName) spans) `shouldBe` 1++  it "FifoHeads blocks a successor behind a delayed head while other groups run" $ \connStr -> do+    observed <- newIORef ([] :: [Text])+    let job = (mkJob "keiro_pgmq_test.fifo_heads_delay") {jobOrdering = FifoHeads}+        tuning =+          withOrdering FifoHeads $+            either (error . show) id $+              mkJobTuning 30 50 (PollEvery 1)+        handler _ctx payload = do+          liftIO $ modifyIORef' observed (<> [payload.message])+          pure Done+    independentDrain <-+      runDb connStr $ do+        ensureOrderedJobQueue job+        _ <- enqueueToGroup job "a" (Ping "a1" 1)+        _ <- runJobOnceWithContext tuning 1 job handler+        _ <- enqueueToGroupWithDelay job 2 "a" (Ping "a2" 2)+        _ <- enqueueToGroup job "a" (Ping "a3" 3)+        _ <- enqueueToGroup job "b" (Ping "b1" 1)+        runJobOnceWithContext tuning 3 job handler+    independentDrain `shouldBe` 1+    readIORef observed `shouldReturn` ["a1", "b1"]+    threadDelay 2_100_000+    recovered <- runDb connStr $ runJobOnceWithContext tuning 2 job handler+    recovered `shouldBe` 2+    readIORef observed `shouldReturn` ["a1", "b1", "a2", "a3"]++  it "FifoHeads advances a group after its head is dead-lettered" $ \connStr -> do+    observed <- newIORef ([] :: [Text])+    let job = (mkJob "keiro_pgmq_test.fifo_heads_dead") {jobOrdering = FifoHeads}+        tuning = withOrdering FifoHeads defaultJobTuning+        record outcome _ctx payload = do+          liftIO $ modifyIORef' observed (<> [payload.message])+          pure outcome+    (deadCount, successorCount, mainDepth, dlqDepth) <-+      runDb connStr $ do+        ensureOrderedJobQueue job+        _ <- enqueueToGroup job "a" (Ping "a1" 1)+        _ <- enqueueToGroup job "a" (Ping "a2" 2)+        deadCount <- runJobOnceWithContext tuning 1 job (record (Dead "poison"))+        successorCount <- runJobOnceWithContext tuning 1 job (record Done)+        mainDepth <- queueLen job.jobQueue.physicalName+        dlqDepth <- queueLen job.jobQueue.dlqName+        pure (deadCount, successorCount, mainDepth, dlqDepth)+    deadCount `shouldBe` 1+    successorCount `shouldBe` 1+    mainDepth `shouldBe` 0+    dlqDepth `shouldBe` 1+    readIORef observed `shouldReturn` ["a1", "a2"]++  it "default one-shot tuning inherits the job's grouped-head ordering" $ \connStr -> do+    let job = (mkJob "keiro_pgmq_test.default_heads") {jobOrdering = FifoHeads}+    len <-+      runDb connStr $ do+        ensureOrderedJobQueue job+        _ <- enqueueToGroup job "a" (Ping "a1" 1)+        runJobOnce 1 job (\_payload -> pure Done)+        queueLen job.jobQueue.physicalName+    len `shouldBe` 0++  it "default worker tuning inherits the job's grouped-head ordering" $ \connStr -> do+    let job = (mkJob "keiro_pgmq_test.default_heads_worker") {jobOrdering = FifoHeads}+    result <-+      try @JobConsumptionConfigError $+        runDb connStr $+          jobProcessor job (\_payload -> pure Done)+    case result of+      Left err -> expectationFailure ("default worker tuning was rejected: " <> show err)+      Right _ -> pure ()++  it "legacy FIFO strategies remain available with batch size one" $ \connStr -> do+    traverse_+      ( \(suffix, declaredOrdering) -> do+          let job = (mkJob ("keiro_pgmq_test.legacy_fifo_" <> suffix)) {jobOrdering = declaredOrdering}+              tuning = withOrdering declaredOrdering defaultJobTuning+          drained <-+            runDb connStr $ do+              ensureOrderedJobQueue job+              _ <- enqueueToGroup job "a" (Ping "a1" 1)+              runJobOnceWithContext tuning 1 job \_ctx _payload -> pure Done+          drained `shouldBe` 1+      )+      [("throughput", FifoThroughput), ("round_robin", FifoRoundRobin)]+   -- EP-3 M4: end-to-end ordering proof.   it "FifoThroughput drain preserves strict within-group order and fully drains" $ \connStr -> do     observed <- newIORef ([] :: [Text])-    let job = mkJob "keiro_pgmq_test.fifo_order"+    let job = (mkJob "keiro_pgmq_test.fifo_order") {jobOrdering = FifoThroughput}     drained <-       runDb connStr $ do         ensureOrderedJobQueue job@@ -1144,13 +1541,13 @@     filter (Text.isPrefixOf "a") log' `shouldBe` ["a1", "a2", "a3"]     filter (Text.isPrefixOf "b") log' `shouldBe` ["b1", "b2"] -  it "FifoThroughput worker path preserves within-group order" $ \connStr -> do+  it "FifoHeads worker path preserves within-group order with a large batch" $ \connStr -> do     observed <- newIORef ([] :: [Text])-    let job = mkJob "keiro_pgmq_test.fifo_worker"+    let job = (mkJob "keiro_pgmq_test.fifo_worker") {jobOrdering = FifoHeads}         tuning =-          withOrdering FifoThroughput $+          withOrdering FifoHeads $             either (error . show) id $-              mkJobTuning 30 1 (PollEvery 0.1)+              mkJobTuning 30 50 (PollEvery 0.1)     processed <-       runDb connStr $ do         ensureOrderedJobQueue job@@ -1177,6 +1574,49 @@     len <- runDb connStr (queueLen job.jobQueue.physicalName)     len `shouldBe` 0 +  it "FifoHeads worker retries a failed head before its successor while other groups progress" $ \connStr -> do+    firstA <- newIORef True+    observed <- newIORef ([] :: [Text])+    let job = (mkJob "keiro_pgmq_test.fifo_worker_retry") {jobOrdering = FifoHeads}+        tuning =+          withOrdering FifoHeads $+            either (error . show) id $+              mkJobTuning 30 50 (PollEvery 0.1)+        handler _ctx payload = do+          liftIO $ modifyIORef' observed (<> [payload.message])+          retryA <- liftIO $ do+            retryPending <- readIORef firstA+            if payload.message == "a1" && retryPending+              then writeIORef firstA False >> pure True+              else pure False+          pure $+            if retryA+              then Retry (RetryDelay 0)+              else Done+    processed <-+      runDb connStr $ do+        ensureOrderedJobQueue job+        _ <- enqueueToGroup job "a" (Ping "a1" 1)+        _ <- enqueueToGroup job "a" (Ping "a2" 2)+        _ <- enqueueToGroup job "b" (Ping "b1" 1)+        result <- runJobWorkers IgnoreFailures 16 [jobProcessorWithContext tuning job handler]+        case result of+          Left err -> liftIO $ fail ("runJobWorkers failed: " <> show err)+          Right app -> do+            ok <- liftIO $ waitUntil do+              messages <- readIORef observed+              pure $+                filter (Text.isPrefixOf "a") messages == ["a1", "a1", "a2"]+                  && filter (Text.isPrefixOf "b") messages == ["b1"]+            stopAppQuickly app+            pure ok+    processed `shouldBe` True+    log' <- readIORef observed+    filter (Text.isPrefixOf "a") log' `shouldBe` ["a1", "a1", "a2"]+    filter (Text.isPrefixOf "b") log' `shouldBe` ["b1"]+    len <- runDb connStr (queueLen job.jobQueue.physicalName)+    len `shouldBe` 0+   -- EP-4 M1: typed metrics surface (main + DLQ).   it "jobQueueMetrics reports main-queue depth after enqueue" $ \connStr -> do     let job = mkJob "keiro_pgmq_test.metrics_depth"@@ -1191,6 +1631,7 @@         pure (mainMetrics, dlqMetrics)     mainMetrics.queueLength `shouldBe` 3     mainMetrics.queueVisibleLength `shouldBe` 3+    mainMetrics.defaultPartitionLength `shouldBe` Nothing     dlqMetrics.queueLength `shouldBe` 0    it "jobDlqMetrics reports DLQ depth after a Dead outcome" $ \connStr -> do@@ -1222,18 +1663,68 @@     retained <- archiveCount connStr (queueNameToText job.jobQueue.dlqName)     retained `shouldBe` 1 +  it "archiveDlqEntries archives inspected rows immediately and is idempotent" $ \connStr -> do+    let job = mkJob "keiro_pgmq_test.dlq_archive_entries"+        unknownId = Pgmq.MessageId 999999+    (entryIds, countArchived, moved, repeated, unknown, remaining) <-+      runDb connStr $ do+        ensureJobQueue job+        _ <- enqueue job (Ping "first" 1)+        _ <- enqueue job (Ping "second" 2)+        runJobOnce 2 job (\_ -> pure (Dead "bad"))+        entries <- readDlq job 2+        let entryIds = map (.dlqMessageId) entries+        countArchived <- archiveDlq job 10+        empty <- archiveDlqEntries job []+        liftIO (empty `shouldBe` [])+        moved <- archiveDlqEntries job entryIds+        repeated <- archiveDlqEntries job entryIds+        unknown <- archiveDlqEntries job [unknownId]+        remaining <- queueLen job.jobQueue.dlqName+        pure (entryIds, countArchived, moved, repeated, unknown, remaining)+    length entryIds `shouldBe` 2+    countArchived `shouldBe` 0+    moved `shouldMatchList` entryIds+    repeated `shouldBe` []+    unknown `shouldBe` []+    remaining `shouldBe` 0+    retained <- archiveCount connStr (queueNameToText job.jobQueue.dlqName)+    retained `shouldBe` 2++  it "inspect archive-by-ids and guarded purge completes without waiting" $ \connStr -> do+    let job = mkJob "keiro_pgmq_test.dlq_no_wait_runbook"+    (inspectedCount, inspectedIds, archivedIds, purgeResult, remaining) <-+      runDb connStr $ do+        ensureJobQueue job+        _ <- enqueue job (Ping "first" 1)+        _ <- enqueue job (Ping "second" 2)+        runJobOnce 2 job (\_ -> pure (Dead "bad"))+        entries <- readDlq job 2+        let inspectedIds = map (.dlqMessageId) entries+        archivedIds <- archiveDlqEntries job inspectedIds+        purgeResult <- purgeDlq job+        remaining <- queueLen job.jobQueue.dlqName+        pure (length entries, inspectedIds, archivedIds, purgeResult, remaining)+    inspectedCount `shouldBe` 2+    archivedIds `shouldMatchList` inspectedIds+    purgeResult `shouldBe` PurgeDlqPurged 0+    remaining `shouldBe` 0+    retained <- archiveCount connStr (queueNameToText job.jobQueue.dlqName)+    retained `shouldBe` 2+   -- EP-4 M3: end-to-end retention lifecycle.   it "archived DLQ rows survive a purge" $ \connStr -> do     let job = mkJob "keiro_pgmq_test.dlq_archive_purge"-    archived <-+    (archived, purged) <-       runDb connStr $ do         ensureJobQueue job         _ <- enqueue job (Ping "poison" 1)         runJobOnce 1 job (\_ -> pure (Dead "bad"))         archived <- archiveDlq job 10-        purgeDlq job-        pure archived+        purged <- purgeDlq job+        pure (archived, purged)     archived `shouldBe` 1+    purged `shouldBe` PurgeDlqPurged 0     dlqLen <- runDb connStr (queueLen job.jobQueue.dlqName)     dlqLen `shouldBe` 0     retained <- archiveCount connStr (queueNameToText job.jobQueue.dlqName)