shibuya-pgmq-adapter 0.16.1.0 → 0.18.0.0

raw patch · 12 files changed

+1177/−24 lines, 12 filesdep +directorydep +hs-opentelemetry-apidep +hs-opentelemetry-exporter-in-memorydep ~ephemeral-pgdep ~pgmq-coredep ~pgmq-effectfulPVP ok

version bump matches the API change (PVP)

Dependencies added: directory, hs-opentelemetry-api, hs-opentelemetry-exporter-in-memory, unix

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

API changes (from Hackage documentation)

Files

• CHANGELOG.md +36/−0
• shibuya-pgmq-adapter.cabal +21/−11
• src/Shibuya/Adapter/Pgmq/Convert.hs +11/−0
• src/Shibuya/Adapter/Pgmq/DeadLetter.hs +39/−0
• src/Shibuya/Adapter/Pgmq/DeadLetter/Internal.hs +354/−0
• src/Shibuya/Adapter/Pgmq/Internal.hs +5/−4
• test/DeadLetterExample.hs +14/−0
• test/Main.hs +2/−0
• test/Shibuya/Adapter/Pgmq/ChaosSpec.hs +79/−5
• test/Shibuya/Adapter/Pgmq/ConvertSpec.hs +38/−0
• test/Shibuya/Adapter/Pgmq/DeadLetterSpec.hs +549/−0
• test/TmpPostgres.hs +29/−4

@@ -1,5 +1,41 @@ # Changelog +## 0.18.0.0 — 2026-10-27++### Breaking Changes++- Require [`shibuya-core`](mori://shinzui/shibuya/packages/shibuya-core) `^>=0.12.0.0`,+  adopting its extended configuration, processor metrics, paused-state API, and+  `Ingested.deliveryStatus` field. PGMQ deliveries report `DeliveryCurrent`,+  preserving handler and acknowledgement behavior.+- Require the [`pgmq-hs`](mori://shinzui/pgmq-hs) 0.7.0.0 package family+  (`pgmq-core`, `pgmq-hasql`, and `pgmq-effectful`; tests also use `pgmq-migration`).++### New Features++- Expose `Shibuya.Adapter.Pgmq.DeadLetter` for explicitly registered,+  non-destructive DLQ browse/fetch and default-disabled transactional redrive.+  Redrive authorizes the source queue and atomically claims each DLQ row before+  sending its original payload; concurrent calls and uncertain-commit retries+  cannot duplicate that physical row's redrive.+- Add `mkDlqPayloadForQueue` and include `original_queue` in adapter-produced+  DLQ envelopes in both metadata modes. Preserve `mkDlqPayload` and existing+  reason fields; historical and malformed envelopes remain inspectable.++### Other Changes++- Use the published dependency releases without the transitional `dlq-admin` flag.+- Extend PostgreSQL coverage for managed DLQs and consumer trace context;+  preserve core configuration defaults in lifecycle fixtures and use stable+  per-user temporary roots for ephemeral PostgreSQL cleanup.+- Add reproducible worker, administration, and encoding performance comparisons.+  Performance acceptance remains incomplete: 196 of 198 numeric gates pass;+  FIFO acknowledgement p95 upper bounds are 1.106857 against the original+  control and 1.172465 against the matched control, exceeding the 1.10 limit.+- Defer performance acceptance for this release by explicit user approval;+  retain the failed bounds and unchanged numeric limits.+- Consolidate the unpublished 0.17.0.0 changes into this release.+ ## 0.16.1.0 — 2026-09-21  ### Reliability

@@ -1,6 +1,6 @@ cabal-version: 3.12 name: shibuya-pgmq-adapter-version: 0.16.1.0+version: 0.18.0.0 synopsis: PGMQ adapter for the Shibuya queue processing framework description:   A Shibuya adapter that integrates with pgmq (PostgreSQL Message Queue)@@ -23,8 +23,10 @@     Shibuya.Adapter.Pgmq     Shibuya.Adapter.Pgmq.Config     Shibuya.Adapter.Pgmq.Convert+    Shibuya.Adapter.Pgmq.DeadLetter    other-modules:+    Shibuya.Adapter.Pgmq.DeadLetter.Internal     Shibuya.Adapter.Pgmq.Internal    default-extensions:@@ -46,10 +48,10 @@     hasql ^>=1.10,     hasql-pool ^>=1.4,     hasql-transaction ^>=1.2,-    pgmq-core ^>=0.6,-    pgmq-effectful ^>=0.6,-    pgmq-hasql ^>=0.6,-    shibuya-core ^>=0.10.0.0,+    pgmq-core ^>=0.7.0.0,+    pgmq-effectful ^>=0.7.0.0,+    pgmq-hasql ^>=0.7.0.0,+    shibuya-core ^>=0.12.0.0,     stm ^>=2.5,     streamly ^>=0.11,     streamly-core ^>=0.3,@@ -75,6 +77,12 @@     src    main-is: Main.hs+  other-modules:+    DeadLetterExample+    Shibuya.Adapter.Pgmq.DeadLetter+    Shibuya.Adapter.Pgmq.DeadLetter.Internal+    Shibuya.Adapter.Pgmq.DeadLetterSpec+   default-extensions:     DeriveAnyClass     DerivingStrategies@@ -107,19 +115,21 @@     base ^>=4.21.0.0,     bytestring,     effectful-core (>=2.6.1 && <2.7) || (>=2.7.1.1 && <2.8),-    ephemeral-pg,+    ephemeral-pg >=0.3.1 && <0.4, directory, unix,     hasql ^>=1.10,     hasql-pool ^>=1.4,     hasql-transaction ^>=1.2,     hspec ^>=2.11,+    hs-opentelemetry-api ^>=1.0,+    hs-opentelemetry-exporter-in-memory ^>=1.0,     pg-migrate >=1.1 && <1.3,-    pgmq-core ^>=0.6,-    pgmq-effectful ^>=0.6,-    pgmq-hasql ^>=0.6,-    pgmq-migration ^>=0.6,+    pgmq-core ^>=0.7.0.0,+    pgmq-effectful ^>=0.7.0.0,+    pgmq-hasql ^>=0.7.0.0,+    pgmq-migration ^>=0.7.0.0,     quickcheck-instances ^>=0.3,     random,-    shibuya-core ^>=0.10.0.0,+    shibuya-core ^>=0.12.0.0,     shibuya-pgmq-adapter,     stm,     streamly ^>=0.11,

@@ -13,6 +13,7 @@      -- * DLQ Payload     mkDlqPayload,+    mkDlqPayloadForQueue,   ) where @@ -169,3 +170,13 @@             "original_headers" .= msg.headers           ]       | otherwise = []++-- | Create an adapter-managed DLQ envelope with an authoritative source queue.+-- The original helper remains byte-for-byte compatible; source identity is+-- present independently of the optional delivery metadata.+mkDlqPayloadForQueue :: Pgmq.QueueName -> Pgmq.Message -> DeadLetterReason -> Bool -> Pgmq.MessageBody+mkDlqPayloadForQueue queue msg reason includeMetadata =+  case mkDlqPayload msg reason includeMetadata of+    Pgmq.MessageBody (Object fields) ->+      Pgmq.MessageBody (Object (KeyMap.insert "original_queue" (String (Pgmq.queueNameToText queue)) fields))+    body -> body

@@ -0,0 +1,39 @@+-- | In-process administration for explicitly registered adapter-managed DLQs.+--+-- Supply a separate administrative pool connected to the workers' database.+-- Constructing a handle performs no I/O and starts no background work. The host+-- owns and closes the pool. Separate connections do not isolate database CPU,+-- I/O or locks. Host transports must enforce their own caller permissions;+-- 'redriveEnabled' is a mutation gate, not authentication.+--+-- Pages are moving views ordered by queue-local message ID. Restart at the first+-- page to discover lower IDs committed late. Limits bound rows, not JSON bytes.+-- Archive recovery and queue discovery/creation are outside this API.+module Shibuya.Adapter.Pgmq.DeadLetter+  ( DeadLetterAdmin,+    DeadLetterAdminConfig (..),+    defaultDeadLetterAdminConfig,+    DlqRegistration (..),+    registrationFromAdapterConfig,+    DlqConfigError (..),+    mkDeadLetterAdmin,+    DlqCursor (..),+    DlqPageRequest (..),+    defaultDlqPageRequest,+    DeadLetterPage (..),+    DeadLetterItem (..),+    DeadLetterEntry (..),+    ReasonRepresentation (..),+    SourceResolution (..),+    DlqDecodeError (..),+    DlqDecodeProblem (..),+    DlqPageError (..),+    DlqError (..),+    RedriveOutcome (..),+    browseDeadLetters,+    fetchDeadLetter,+    redriveDeadLetter,+  )+where++import Shibuya.Adapter.Pgmq.DeadLetter.Internal

@@ -0,0 +1,354 @@+-- | Implementation of adapter-owned DLQ semantics and transaction composition.+module Shibuya.Adapter.Pgmq.DeadLetter.Internal where++import Control.Monad (unless, void, when)+import Data.Aeson (FromJSON, Result (..), Value (..), fromJSON)+import Data.Aeson.Key qualified as Key+import Data.Aeson.KeyMap qualified as KM+import Data.Functor.Contravariant (contramap)+import Data.Int (Int32, Int64)+import Data.List (nub)+import Data.Text (Text)+import Data.Text qualified as Text+import Data.Time (UTCTime)+import Data.Vector (Vector)+import Data.Vector qualified as Vector+import Hasql.Decoders qualified as D+import Hasql.Encoders qualified as E+import Hasql.Pool qualified as Pool+import Hasql.Statement qualified as Statement+import Hasql.Transaction qualified as Tx+import Hasql.Transaction.Sessions qualified as TxSession+import Pgmq.Effectful (PgmqRuntimeError, fromUsageError)+import Pgmq.Hasql.Decoders (messageDecoder)+import Pgmq.Hasql.Encoders (messageIdValue)+import Pgmq.Hasql.Statements.Inspection qualified as Inspection+import Pgmq.Hasql.Statements.Message qualified as Msg+import Pgmq.Hasql.Statements.Types qualified as Q+import Pgmq.Types (Message (..), MessageBody (..), MessageHeaders (..), MessageId (..), QueueName, parseQueueName, queueNameToText)+import Shibuya.Adapter.Pgmq.Config (DeadLetterConfig (..), DeadLetterTarget (..), PgmqAdapterConfig (..))++-- | One explicitly managed physical DLQ and the sources it may return to.+data DlqRegistration = DlqRegistration+  { deadLetterQueue :: !QueueName,+    allowedSourceQueues :: ![QueueName]+  }+  deriving stock (Eq, Show)++-- | Administrative settings, independent of worker configuration.+data DeadLetterAdminConfig = DeadLetterAdminConfig+  { registrations :: ![DlqRegistration],+    redriveEnabled :: !Bool,+    statementTimeoutMs :: !Int32,+    lockTimeoutMs :: !Int32+  }+  deriving stock (Eq, Show)++-- | Disable mutation; bound each statement to 5 seconds and lock waits to 1.+defaultDeadLetterAdminConfig :: [DlqRegistration] -> DeadLetterAdminConfig+defaultDeadLetterAdminConfig rs = DeadLetterAdminConfig rs False 5000 1000++-- | Invalid configuration is rejected without acquiring a connection.+data DlqConfigError+  = DuplicateManagedQueue !QueueName+  | EmptySourceQueues !QueueName+  | DuplicateSourceQueues !QueueName+  | SelfMapping !QueueName+  | InvalidStatementTimeout !Int32+  | InvalidLockTimeout !Int32+  | NoDirectDeadLetterTarget+  deriving stock (Eq, Show)++-- | Borrowed pool and validated configuration; no threads, queries or ownership.+data DeadLetterAdmin = DeadLetterAdmin !Pool.Pool !DeadLetterAdminConfig++-- | Validate registrations and deadlines without performing I/O. The supplied+-- pool is borrowed; the host remains responsible for its lifetime.+mkDeadLetterAdmin :: Pool.Pool -> DeadLetterAdminConfig -> Either DlqConfigError DeadLetterAdmin+mkDeadLetterAdmin pool config = do+  when (config.statementTimeoutMs <= 0) $ Left (InvalidStatementTimeout config.statementTimeoutMs)+  when (config.lockTimeoutMs <= 0) $ Left (InvalidLockTimeout config.lockTimeoutMs)+  let validate seen registration = do+        let queue = registration.deadLetterQueue+            sources = registration.allowedSourceQueues+        when (queue `elem` seen) $ Left (DuplicateManagedQueue queue)+        when (null sources) $ Left (EmptySourceQueues queue)+        when (sources /= nub sources) $ Left (DuplicateSourceQueues queue)+        when (queue `elem` sources) $ Left (SelfMapping queue)+        pure (queue : seen)+      go _ [] = pure ()+      go seen (r : rs) = validate seen r >>= \next -> go next rs+  go [] config.registrations+  pure (DeadLetterAdmin pool config)++-- | Derive the single-source registration for a direct adapter target.+-- Topic destinations require explicit physical registrations.+registrationFromAdapterConfig :: PgmqAdapterConfig -> Either DlqConfigError DlqRegistration+registrationFromAdapterConfig config = case config.deadLetterConfig of+  Just DeadLetterConfig {dlqTarget = DirectQueue queue}+    | queue /= config.queueName -> Right (DlqRegistration queue [config.queueName])+    | otherwise -> Left (SelfMapping queue)+  _ -> Left NoDirectDeadLetterTarget++-- | Exclusive cursor. IDs have meaning only within this physical queue's lifetime.+data DlqCursor = DlqCursor+  { deadLetterQueue :: !QueueName,+    afterMessageId :: !MessageId+  }+  deriving stock (Eq, Show)++-- | Positive row limit (at most 500), with an optional exclusive cursor.+data DlqPageRequest = DlqPageRequest+  { pageLimit :: !Int32,+    afterCursor :: !(Maybe DlqCursor)+  }+  deriving stock (Eq, Show)++-- | Start at the first ID and return at most 50 rows.+defaultDlqPageRequest :: DlqPageRequest+defaultDlqPageRequest = DlqPageRequest 50 Nothing++-- | Whether reason fields were stored explicitly or inferred from old text.+data ReasonRepresentation = StructuredReason | LegacyReason+  deriving stock (Eq, Show)++-- | How the source was resolved. A multi-source historical row remains unknown.+data SourceResolution = PayloadSource | ConfiguredFallback | SourceUnavailable+  deriving stock (Eq, Show)++-- | Stable, field-specific reasons an envelope cannot be interpreted safely.+data DlqDecodeProblem+  = MissingField+  | WrongType !Text+  | InvalidSourceName !Text+  | SourceNotAllowed !QueueName+  deriving stock (Eq, Show)++-- | The envelope field that could not be interpreted and its typed problem.+data DlqDecodeError = DlqDecodeError+  { field :: !Text,+    problem :: !DlqDecodeProblem+  }+  deriving stock (Eq, Show)++-- | Original delivery metadata and current DLQ metadata are deliberately separate.+-- Missing 'originalHeaders' is 'Nothing'; explicit JSON null is @Just Null@.+data DeadLetterEntry = DeadLetterEntry+  { originalPayload :: !Value,+    reasonCode :: !Text,+    reasonDetail :: !Value,+    reasonRepresentation :: !ReasonRepresentation,+    resolvedSource :: !(Maybe QueueName),+    sourceResolution :: !SourceResolution,+    originalMessageId :: !(Maybe MessageId),+    originalReadCount :: !(Maybe Int64),+    originalEnqueuedAt :: !(Maybe UTCTime),+    originalLastReadAt :: !(Maybe UTCTime),+    originalHeaders :: !(Maybe Value),+    dlqMessageId :: !MessageId,+    dlqReadCount :: !Int64,+    dlqEnqueuedAt :: !UTCTime,+    dlqLastReadAt :: !(Maybe UTCTime)+  }+  deriving stock (Eq, Show)++-- | Corrupt envelopes stay in their original position and retain their raw row.+data DeadLetterItem = DeadLetterItem+  { deadLetterQueue :: !QueueName,+    rawMessage :: !Message,+    decodedEntry :: !(Either DlqDecodeError DeadLetterEntry)+  }+  deriving stock (Eq, Show)++-- | A moving view, not a shared snapshot. Refresh page one for late lower IDs.+-- The row limit does not bound JSON bytes or database scan work.+data DeadLetterPage = DeadLetterPage+  { items :: !(Vector DeadLetterItem),+    nextCursor :: !(Maybe DlqCursor)+  }+  deriving stock (Eq, Show)++-- | Invalid row limit or a cursor bound to a different physical DLQ.+data DlqPageError = InvalidPageLimit !Int32 | CursorQueueMismatch !QueueName !QueueName+  deriving stock (Eq, Show)++-- | Database failures never become empty pages or successful absence.+data DlqError+  = RedriveDisabled+  | UnknownManagedDlq !QueueName+  | InvalidPageRequest !DlqPageError+  | EntryNotFound !QueueName !MessageId+  | MalformedEnvelope !DlqDecodeError+  | UnresolvedSource+  | DisallowedSource !QueueName+  | DatabaseError !PgmqRuntimeError+  deriving stock (Eq, Show)++-- | Absence cannot establish which actor removed the row.+data RedriveOutcome = Redriven !MessageId | AlreadyAbsent+  deriving stock (Eq, Show)++findRegistration :: DeadLetterAdminConfig -> QueueName -> Either DlqError DlqRegistration+findRegistration config queue = case filter (\r -> r.deadLetterQueue == queue) config.registrations of+  [registration] -> Right registration+  _ -> Left (UnknownManagedDlq queue)++decodeEntry :: DlqRegistration -> Message -> Either DlqDecodeError DeadLetterEntry+decodeEntry registration msg = case unMessageBody msg.body of+  Object fields -> do+    originalPayload <- required fields "original_message"+    (reasonCode, reasonDetail, reasonRepresentation) <- decodeReason fields+    (resolvedSource, sourceResolution) <- case KM.lookup "original_queue" fields of+      Nothing -> pure $ case registration.allowedSourceQueues of+        [source] -> (Just source, ConfiguredFallback)+        _ -> (Nothing, SourceUnavailable)+      Just value -> do+        name <- typed "original_queue" value+        source <- either (const $ bad "original_queue" (InvalidSourceName name)) pure (parseQueueName name)+        unless (source `elem` registration.allowedSourceQueues) $ bad "original_queue" (SourceNotAllowed source)+        pure (Just source, PayloadSource)+    originalMessageId <- fmap MessageId <$> optional fields "original_message_id"+    originalReadCount <- optional fields "read_count"+    originalEnqueuedAt <- optional fields "original_enqueued_at"+    originalLastReadAt <- case KM.lookup "last_read_at" fields of+      Nothing -> pure Nothing+      Just Null -> pure Nothing+      Just value -> Just <$> typed "last_read_at" value+    let originalHeaders = KM.lookup "original_headers" fields+    pure+      DeadLetterEntry+        { originalPayload,+          reasonCode,+          reasonDetail,+          reasonRepresentation,+          resolvedSource,+          sourceResolution,+          originalMessageId,+          originalReadCount,+          originalEnqueuedAt,+          originalLastReadAt,+          originalHeaders,+          dlqMessageId = msg.messageId,+          dlqReadCount = msg.readCount,+          dlqEnqueuedAt = msg.enqueuedAt,+          dlqLastReadAt = msg.lastReadAt+        }+  _ -> bad "$" (WrongType "object")+  where+    decodeReason fields+      | KM.member "dead_letter_reason_code" fields || KM.member "dead_letter_reason_detail" fields = do+          code <- required fields "dead_letter_reason_code" >>= typed "dead_letter_reason_code"+          detail <- required fields "dead_letter_reason_detail"+          pure (code, detail, StructuredReason)+      | otherwise = do+          rendered <- required fields "dead_letter_reason" >>= typed "dead_letter_reason"+          let (code, suffix) = Text.breakOn ": " rendered+          pure (code, if Text.null suffix then Null else String (Text.drop 2 suffix), LegacyReason)+    required fields key = maybe (bad (Key.toText key) MissingField) pure (KM.lookup key fields)+    optional fields key = traverse (typed (Key.toText key)) (KM.lookup key fields)++bad :: Text -> DlqDecodeProblem -> Either DlqDecodeError a+bad field problem = Left DlqDecodeError {field, problem}++typed :: (FromJSON a) => Text -> Value -> Either DlqDecodeError a+typed field value = case fromJSON value of+  Error err -> bad field (WrongType (Text.pack err))+  Success result -> Right result++inspectItem :: DlqRegistration -> Message -> DeadLetterItem+inspectItem registration msg = DeadLetterItem registration.deadLetterQueue msg (decodeEntry registration msg)++-- | Transaction-local settings are reset by PostgreSQL at commit or rollback.+setTimeouts :: DeadLetterAdminConfig -> Tx.Transaction ()+setTimeouts config =+  void $+    Tx.statement+      (Text.pack (show config.statementTimeoutMs) <> "ms", Text.pack (show config.lockTimeoutMs) <> "ms")+      ( Statement.preparable+          "select set_config('statement_timeout', $1, true), set_config('lock_timeout', $2, true)"+          (contramap fst (E.param (E.nonNullable E.text)) <> contramap snd (E.param (E.nonNullable E.text)))+          (D.singleRow ((,) <$> D.column (D.nonNullable D.text) <*> D.column (D.nonNullable D.text)))+      )++runTransaction :: DeadLetterAdmin -> TxSession.Mode -> Tx.Transaction (Either DlqError a) -> IO (Either DlqError a)+runTransaction (DeadLetterAdmin pool config) mode tx = do+  result <- Pool.use pool $ TxSession.transactionNoRetry TxSession.ReadCommitted mode (setTimeouts config >> tx)+  pure $ either (Left . DatabaseError . fromUsageError) id result++resolveTable :: QueueName -> Tx.Transaction Text+resolveTable queue = Tx.statement (queueNameToText queue, "q") Inspection.formatTableName++-- | Observe rows without leasing, including invisible and malformed rows.+browseDeadLetters :: DeadLetterAdmin -> QueueName -> DlqPageRequest -> IO (Either DlqError DeadLetterPage)+browseDeadLetters admin@(DeadLetterAdmin _ config) queue request = case validate of+  Left err -> pure (Left err)+  Right registration -> runTransaction admin TxSession.Read $ do+    table <- resolveTable queue+    rows <- Tx.statement (fmap (\cursor -> cursor.afterMessageId) request.afterCursor, request.pageLimit + 1) (Inspection.peekStatement table)+    let page = Vector.take (fromIntegral request.pageLimit) rows+        cursor =+          if Vector.length rows > fromIntegral request.pageLimit+            then (\msg -> DlqCursor queue msg.messageId) <$> (page Vector.!? (Vector.length page - 1))+            else Nothing+    pure $ Right (DeadLetterPage (Vector.map (inspectItem registration) page) cursor)+  where+    validate = do+      registration <- findRegistration config queue+      when (request.pageLimit < 1 || request.pageLimit > 500) $+        Left (InvalidPageRequest (InvalidPageLimit request.pageLimit))+      case request.afterCursor of+        Just cursor | cursor.deadLetterQueue /= queue -> Left (InvalidPageRequest (CursorQueueMismatch queue cursor.deadLetterQueue))+        _ -> pure ()+      pure registration++-- | Observe one registered DLQ row without leasing it. Malformed rows are+-- returned as inspectable items; only actual absence yields EntryNotFound.+fetchDeadLetter :: DeadLetterAdmin -> QueueName -> MessageId -> IO (Either DlqError DeadLetterItem)+fetchDeadLetter admin@(DeadLetterAdmin _ config) queue messageId = case findRegistration config queue of+  Left err -> pure (Left err)+  Right registration -> runTransaction admin TxSession.Read $ do+    table <- resolveTable queue+    result <- Tx.statement messageId (Inspection.lookupStatement table)+    pure $ maybe (Left (EntryNotFound queue messageId)) (Right . inspectItem registration) result++-- | Serialize on the selected row and atomically delete/send its original body.+-- Retry the complete operation after uncertain commit. A new ID/count/time and+-- tail position are assigned; distinct fan-out copies have distinct claims.+redriveDeadLetter :: DeadLetterAdmin -> QueueName -> MessageId -> IO (Either DlqError RedriveOutcome)+redriveDeadLetter admin@(DeadLetterAdmin _ config) queue messageId+  | not config.redriveEnabled = pure (Left RedriveDisabled)+  | otherwise = case findRegistration config queue of+      Left err -> pure (Left err)+      Right registration -> runTransaction admin TxSession.Write $ do+        table <- resolveTable queue+        locked <- Tx.statement messageId (lockedLookup table)+        case locked of+          Nothing -> pure (Right AlreadyAbsent)+          Just msg -> case decodeEntry registration msg of+            Left err -> pure $ Left $ case err.problem of+              SourceNotAllowed source -> DisallowedSource source+              _ -> MalformedEnvelope err+            Right entry -> case entry.resolvedSource of+              Nothing -> pure (Left UnresolvedSource)+              Just source -> do+                claimed <- Tx.statement (Q.MessageQuery queue messageId) Msg.deleteMessage+                if not claimed+                  then pure (Right AlreadyAbsent)+                  else do+                    let headers = case entry.originalHeaders of+                          Just saved -> Just saved+                          Nothing -> msg.headers+                    newId <- case headers of+                      Nothing -> Tx.statement (Q.SendMessage source (MessageBody entry.originalPayload) Nothing) Msg.sendMessage+                      Just value -> Tx.statement (Q.SendMessageWithHeaders source (MessageBody entry.originalPayload) (MessageHeaders value) Nothing) Msg.sendMessageWithHeaders+                    pure (Right (Redriven newId))++lockedLookup :: Text -> Statement.Statement MessageId (Maybe Message)+lockedLookup table =+  Statement.unpreparable+    ( "select msg_id, read_ct, enqueued_at, last_read_at, vt, message, headers from pgmq."+        <> Inspection.quoteIdentifier table+        <> " where msg_id = $1 for update"+    )+    (E.param (E.nonNullable messageIdValue))+    (D.rowMaybe messageDecoder)

@@ -107,12 +107,12 @@     PollingConfig (..),   ) import Shibuya.Adapter.Pgmq.Convert-  ( mkDlqPayload,+  ( mkDlqPayloadForQueue,     pgmqMessageToEnvelope,   ) import Shibuya.Core.Ack (AckDecision (..), DeadLetterReason (..), RetryDelay (..)) import Shibuya.Core.AckHandle (AckHandle (..))-import Shibuya.Core.Ingested (Ingested (..))+import Shibuya.Core.Ingested (DeliveryStatus (..), Ingested (..)) import Shibuya.Core.Lease (Lease (..)) import Shibuya.Core.Types (TraceHeaders) import Shibuya.Telemetry.Effect (Tracing)@@ -309,7 +309,7 @@     Left err -> throwError (fromUsageError err)     Right () -> pure ()   where-    dlqBody = mkDlqPayload msg reason dlqConfig.includeMetadata+    dlqBody = mkDlqPayloadForQueue config.queueName msg reason dlqConfig.includeMetadata     sourceQuery = MessageQuery config.queueName msg.messageId     session =       Transaction.Sessions.transaction@@ -446,7 +446,8 @@           Ingested             { envelope = pgmqMessageToEnvelope msg,               ack = ackHandle,-              lease = Just lease+              lease = Just lease,+              deliveryStatus = pure DeliveryCurrent             }  finalizeAutoDeadLetter ::

@@ -0,0 +1,14 @@+module DeadLetterExample (configureOrders, inspectOrders) where++import Hasql.Pool qualified as Pool+import Pgmq.Types (QueueName)+import Shibuya.Adapter.Pgmq.DeadLetter++configureOrders :: Pool.Pool -> QueueName -> QueueName -> Either DlqConfigError DeadLetterAdmin+configureOrders administrativePool source dlq =+  mkDeadLetterAdmin+    administrativePool+    (defaultDeadLetterAdminConfig [DlqRegistration dlq [source]])++inspectOrders :: DeadLetterAdmin -> QueueName -> IO (Either DlqError DeadLetterPage)+inspectOrders admin dlq = browseDeadLetters admin dlq defaultDlqPageRequest

@@ -3,6 +3,7 @@ import Shibuya.Adapter.Pgmq.ChaosSpec qualified as ChaosSpec import Shibuya.Adapter.Pgmq.ConfigSpec qualified as ConfigSpec import Shibuya.Adapter.Pgmq.ConvertSpec qualified as ConvertSpec+import Shibuya.Adapter.Pgmq.DeadLetterSpec qualified as DeadLetterSpec import Shibuya.Adapter.Pgmq.IntegrationSpec qualified as IntegrationSpec import Shibuya.Adapter.Pgmq.InternalSpec qualified as InternalSpec import Shibuya.Adapter.Pgmq.PropertySpec qualified as PropertySpec@@ -18,3 +19,4 @@     describe "Shibuya.Adapter.Pgmq Properties" PropertySpec.spec   describe "Integration Tests" IntegrationSpec.spec   describe "Chaos Tests" ChaosSpec.spec+  DeadLetterSpec.spec

@@ -15,10 +15,12 @@ import Control.Exception qualified as Exception import Control.Monad (forM_, when) import Data.Aeson (Value (..), object, (.=))+import Data.Aeson.Key qualified as Key import Data.Aeson.KeyMap qualified as KeyMap import Data.IORef (IORef, atomicModifyIORef', newIORef, readIORef) import Data.Int (Int32) import Data.Text qualified as Text+import Data.Text.Encoding qualified as Text import Data.Vector qualified as Vector import Effectful (Eff, IOE, Limit (..), Persistence (..), UnliftStrategy (..), liftIO, runEff, withEffToIO, (:>)) import Effectful.Dispatch.Dynamic (interpret)@@ -27,13 +29,17 @@ import Hasql.Pool qualified as Pool import Hasql.Session qualified as Session import Hasql.Statement qualified as Statement+import OpenTelemetry.Attributes (emptyAttributes)+import OpenTelemetry.Exporter.InMemory (inMemoryListExporter)+import OpenTelemetry.Trace.Core qualified as OTel+import OpenTelemetry.Trace.Id.Generator (IdGenerator (..)) import Pgmq.Effectful (Pgmq, PgmqRuntimeError, runPgmq) import Pgmq.Effectful qualified as PgmqEff import Pgmq.Effectful.Effect qualified as PgmqEffect import Pgmq.Hasql.Encoders qualified as Encoders import Pgmq.Hasql.Sessions qualified as Sessions-import Pgmq.Hasql.Statements.Types (ReadMessage (..), SendMessage (..), SendMessageWithHeaders (..))-import Pgmq.Types (Message (..), MessageBody (..), MessageHeaders (..), QueueName, parseQueueName)+import Pgmq.Hasql.Statements.Types (BindTopic (..), ReadMessage (..), SendMessage (..), SendMessageWithHeaders (..))+import Pgmq.Types (Message (..), MessageBody (..), MessageHeaders (..), QueueName, parseQueueName, parseRoutingKey, parseTopicPattern) import Shibuya.Adapter (Adapter (..)) import Shibuya.Adapter.Pgmq   ( PgmqAcknowledgementException,@@ -47,6 +53,7 @@     directDeadLetter,     mkPgmqAdapterEnv,     pgmqAdapter,+    topicDeadLetter,   ) import Shibuya.Adapter.Pgmq.Internal (mkAckHandle, mkIngested) import Shibuya.App@@ -67,7 +74,8 @@ import Shibuya.Core.Ingested (Ingested (..)) import Shibuya.Handler (Handler) import Shibuya.Internal.Runner.Master (getLifecycleSnapshot)-import Shibuya.Telemetry.Effect (Tracing, runTracingNoop)+import Shibuya.Telemetry.Effect (Tracing, runTracing, runTracingNoop)+import Shibuya.Telemetry.Propagation (extractTraceContext) import Streamly.Data.Stream qualified as Stream import System.Environment (lookupEnv) import System.Timeout (timeout)@@ -334,6 +342,72 @@             }     Vector.length sourceMessages `shouldBe` 0 +  forM_ [False, True] $ \topicRoute ->+    forM_ [False, True] $ \withOriginal ->+      it ("DLQ consumer trace " <> (if topicRoute then "topic" else "direct") <> (if withOriginal then " with producer" else " without headers")) $ \TestFixture {pool, queueName, dlqName} -> do+        let producerParent = "00-0af7651916cd43dd8448eb211c80319c-b7ad6b7169203331-01"+            producerState = "vendor=opaque"+            original = object ["traceparent" .= (producerParent :: Text.Text), "tracestate" .= (producerState :: Text.Text), "custom" .= ("kept" :: Text.Text)]+            routingKey = either (error . show) id $ parseRoutingKey "audit.dlq"+            topicPattern = either (error . show) id $ parseTopicPattern "audit.dlq"+            config =+              (defaultConfig queueName)+                { visibilityTimeout = 5,+                  batchSize = 1,+                  polling = StandardPolling {pollInterval = 0.1},+                  deadLetterConfig = Just $ if topicRoute then topicDeadLetter routingKey True else directDeadLetter dlqName True+                }+        when topicRoute $ runPgmqSession pool $ Sessions.bindTopic $ BindTopic topicPattern dlqName+        runPgmqSession pool $ do+          if withOriginal+            then do+              _ <- Sessions.sendMessageWithHeaders $ SendMessageWithHeaders queueName (MessageBody (String "traced-dlq")) (MessageHeaders original) (Just 0)+              pure ()+            else do+              _ <- Sessions.sendMessage $ SendMessage queueName (MessageBody (String "traced-dlq")) (Just 0)+              pure ()+        (spanProcessor, spansRef) <- inMemoryListExporter+        processedRef <- newIORef (0 :: Int)+        Exception.bracket+          (OTel.createTracerProvider [spanProcessor] OTel.emptyTracerProviderOptions {OTel.tracerProviderOptionsIdGenerator = DefaultIdGenerator})+          (\p -> OTel.shutdownTracerProvider p (Just 5_000_000))+          $ \provider -> do+            let tracer = OTel.makeTracer provider (OTel.InstrumentationLibrary "dlq-audit" "" "" emptyAttributes) OTel.tracerOptions+            runAdapterIO pool $ runTracing tracer $ do+              adapter <- requireAdapter pool config+              result <- runApp defaultAppConfig [(ProcessorId "dlq-identity", mkProcessor adapter (deadLetterHandler processedRef))]+              case result of+                Left err -> liftIO $ expectationFailure $ show err+                Right appHandle -> do+                  liftIO $ waitForProcessed processedRef 1 3_000_000+                  drained <- stopAppGracefully defaultShutdownConfig {drainTimeout = 5} appHandle+                  liftIO $ drained `shouldBe` True+        messages <- runPgmqSession pool $ Sessions.readMessage $ ReadMessage dlqName 30 (Just 10) Nothing+        Vector.length messages `shouldBe` 1+        spans <- readIORef spansRef+        named <- mapM (\sp -> do hot <- readIORef (OTel.spanHot sp); pure (OTel.hotName hot, sp)) spans+        case (Vector.uncons messages, [sp | ("dlq-identity process", sp) <- named]) of+          (Just (msg, _), [consumer]) -> case msg.headers of+            Just (Object hdrs) -> do+              let stringHeader k = case KeyMap.lookup k hdrs of+                    Just (String v) -> Just (Text.encodeUtf8 v)+                    _ -> Nothing+                  outgoing = [(Text.encodeUtf8 (Key.toText k), v) | k <- ["traceparent", "tracestate"], Just v <- [stringHeader k]]+              case extractTraceContext outgoing of+                Nothing -> expectationFailure "expected valid consumer context in DLQ"+                Just ctx -> do+                  OTel.traceId ctx `shouldBe` OTel.traceId (OTel.spanContext consumer)+                  OTel.spanId ctx `shouldBe` OTel.spanId (OTel.spanContext consumer)+              if withOriginal+                then do+                  KeyMap.lookup "x-shibuya-upstream-traceparent" hdrs `shouldBe` Just (String producerParent)+                  KeyMap.lookup "x-shibuya-upstream-tracestate" hdrs `shouldBe` Just (String producerState)+                  KeyMap.lookup "custom" hdrs `shouldBe` Just (String "kept")+                  stringHeader "traceparent" `shouldNotBe` Just (Text.encodeUtf8 producerParent)+                else KeyMap.lookup "x-shibuya-upstream-traceparent" hdrs `shouldBe` Nothing+            _ -> expectationFailure "expected DLQ headers"+          _ -> expectationFailure "expected one DLQ message and one consumer span"+   it "preserves trace headers when moving to DLQ" $ \TestFixture {pool, queueName, dlqName} -> do     -- Send a message with trace headers     let traceparent = "00-0af7651916cd43dd8448eb211c80319c-b7ad6b7169203331-01" :: Text.Text@@ -1162,7 +1236,7 @@       adapter <- requireAdapter pool config       let handler = slowHandler processedRef 10_000_000           processor = mkProcessor adapter handler-      appResult <- runApp (AppConfig {strategy = IgnoreFailures, inboxSize = 2}) [(ProcessorId "noloss-test", processor)]+      appResult <- runApp (defaultAppConfig {strategy = IgnoreFailures, inboxSize = 2}) [(ProcessorId "noloss-test", processor)]       case appResult of         Left err -> liftIO $ expectationFailure $ "Failed to start app: " <> show err         Right appHandle -> do@@ -1328,7 +1402,7 @@     adapter <- requireAdapter pool config     let handler = slowHandler processedRef 400000 -- 0.4s per message         processor = mkProcessor adapter handler-    appResult <- runApp (AppConfig {strategy = IgnoreFailures, inboxSize = 2}) [(ProcessorId "shutdown-measure", processor)]+    appResult <- runApp (defaultAppConfig {strategy = IgnoreFailures, inboxSize = 2}) [(ProcessorId "shutdown-measure", processor)]     case appResult of       Left err -> liftIO $ expectationFailure $ "Failed to start app: " <> show err       Right appHandle -> do

@@ -21,6 +21,7 @@   extractTraceHeadersSpec   pgmqMessageToEnvelopeSpec   mkDlqPayloadSpec+  mkDlqPayloadForQueueSpec  -- | Tests for messageIdToShibuya messageIdToShibuyaSpec :: Spec@@ -455,3 +456,40 @@   case mkDeadLetterCode "keiro.router.selection.recipient_overflow" of     Left err -> error (Text.unpack err)     Right code -> code++mkDlqPayloadForQueueSpec :: Spec+mkDlqPayloadForQueueSpec = describe "mkDlqPayloadForQueue" $ do+  let timestamp = UTCTime (fromGregorian 2026 10 25) 0+      queue = either (error . show) id (Pgmq.parseQueueName "orders")+      msg =+        Pgmq.Message+          (Pgmq.MessageId 42)+          timestamp+          timestamp+          (Just timestamp)+          3+          (Pgmq.MessageBody (object ["order" .= (7 :: Int)]))+          (Just (object ["x-pgmq-group" .= ("a" :: Text.Text)]))+      reason = ApplicationFailure representativeDeadLetterCode "注文: failed"+      baseFields =+        [ "original_queue" .= ("orders" :: Text.Text),+          "original_message" .= object ["order" .= (7 :: Int)],+          "dead_letter_reason" .= ("keiro.router.selection.recipient_overflow: 注文: failed" :: Text.Text),+          "dead_letter_reason_code" .= ("keiro.router.selection.recipient_overflow" :: Text.Text),+          "dead_letter_reason_detail" .= ("注文: failed" :: Text.Text)+        ]+  it "writes the exact queue-aware envelope without optional metadata" $+    mkDlqPayloadForQueue queue msg reason False `shouldBe` Pgmq.MessageBody (object baseFields)+  it "writes the exact queue-aware envelope with original metadata" $+    mkDlqPayloadForQueue queue msg reason True+      `shouldBe` Pgmq.MessageBody+        ( object+            ( baseFields+                ++ [ "original_message_id" .= (42 :: Int),+                     "original_enqueued_at" .= timestamp,+                     "last_read_at" .= timestamp,+                     "read_count" .= (3 :: Int),+                     "original_headers" .= object ["x-pgmq-group" .= ("a" :: Text.Text)]+                   ]+            )+        )

@@ -0,0 +1,549 @@+module Shibuya.Adapter.Pgmq.DeadLetterSpec (spec) where++import Control.Concurrent (threadDelay)+import Control.Concurrent.Async (async, cancel, concurrently, waitCatch, withAsync)+import Control.Concurrent.MVar (newEmptyMVar, putMVar, takeMVar)+import Control.Exception (bracket, throwIO)+import Control.Monad (forM_, void)+import Data.Aeson (Value (..), object, (.=))+import Data.Aeson.Key qualified+import Data.Aeson.KeyMap qualified as KM+import Data.Either (isLeft)+import Data.Text (Text)+import Data.Time (secondsToDiffTime)+import Data.Vector qualified as V+import Effectful (liftIO, runEff)+import Effectful.Error.Static (runErrorNoCallStack)+import Hasql.Connection.Settings qualified as Settings+import Hasql.Decoders qualified as D+import Hasql.Encoders qualified as E+import Hasql.Pool qualified as Pool+import Hasql.Pool.Config qualified as PoolConfig+import Hasql.Session qualified as Session+import Hasql.Statement qualified as Statement+import Pgmq.Effectful (PgmqRuntimeError, runPgmq)+import Pgmq.Hasql.Sessions qualified as S+import Pgmq.Hasql.Statements.Inspection qualified as Inspection+import Pgmq.Hasql.Statements.Types qualified as Q+import Pgmq.Types qualified as P+import Shibuya.Adapter.Pgmq+import Shibuya.Adapter.Pgmq.DeadLetter+import Shibuya.App (ProcessorId (..), defaultAppConfig, defaultShutdownConfig, mkProcessor, runApp, stopAppGracefully)+import Shibuya.Core.Ack (AckDecision (..), DeadLetterReason (..), mkDeadLetterCode)+import Shibuya.Core.Ingested (Message (..))+import Shibuya.Core.Types (Envelope (..))+import Shibuya.Telemetry.Effect (runTracingNoop)+import System.Timeout (timeout)+import Test.Hspec hiding (before)+import TmpPostgres++spec :: Spec+spec = describe "Dead-letter administration" $ do+  around database $ do+    it "preserves an application's exact reason through a real handler, browse and fetch" $ \fixture -> do+      let pool = fixture.pool+          source = fixture.queueName+          dlq = fixture.dlqName+          payload = object ["order" .= (42 :: Int)]+          detail = "注文の検証: failed: exact text"+          code = either (error . show) id (mkDeadLetterCode "orders.validation.failed")+          config =+            (defaultConfig source)+              { batchSize = 1,+                polling = StandardPolling 0.01,+                deadLetterConfig = Just (directDeadLetter dlq True)+              }+      void $ runPgmqSession pool $ S.sendMessage (Q.SendMessage source (P.MessageBody payload) Nothing)+      outcome <- runEff $ runErrorNoCallStack @PgmqRuntimeError $ runPgmq pool $ runTracingNoop $ do+        adapter <- pgmqAdapter (mkPgmqAdapterEnv pool) config >>= either (error . show) pure+        app <- runApp defaultAppConfig [(ProcessorId "dlq-roundtrip", mkProcessor adapter (\_ -> pure (AckDeadLetter (ApplicationFailure code detail))))] >>= either (error . show) pure+        liftIO $ awaitDlq pool dlq+        drained <- stopAppGracefully defaultShutdownConfig app+        liftIO $ drained `shouldBe` True+      either throwIO pure outcome+      admin <- requireAdmin fixture False+      page <- browseDeadLetters admin dlq defaultDlqPageRequest >>= right+      V.length page.items `shouldBe` 1+      let item = V.head page.items+      entry <- right item.decodedEntry+      entry.reasonCode `shouldBe` "orders.validation.failed"+      entry.reasonDetail `shouldBe` String detail+      entry.originalPayload `shouldBe` payload+      entry.resolvedSource `shouldBe` Just source+      entry.sourceResolution `shouldBe` PayloadSource+      entry.originalReadCount `shouldBe` Just 1+      entry.dlqReadCount `shouldBe` 0+      entry.dlqEnqueuedAt `shouldBe` item.rawMessage.enqueuedAt+      fetchDeadLetter admin dlq item.rawMessage.messageId `shouldReturn` Right item+      redriveDeadLetter admin dlq item.rawMessage.messageId `shouldReturn` Left RedriveDisabled++    it "pages by exclusive IDs across malformed and invisible rows without changing any row" $ \fixture -> do+      admin <- requireAdmin fixture False+      browseDeadLetters admin fixture.dlqName defaultDlqPageRequest `shouldReturn` Right (DeadLetterPage V.empty Nothing)+      firstId <- seed fixture (envelope (String "first"))+      corruptId <- seed fixture (String "not an envelope")+      lastId <- seed fixture (envelope Null)+      void $ runPgmqSession fixture.pool $ S.readMessage (Q.ReadMessage fixture.dlqName 300 (Just 2) Nothing)+      before <- snapshot fixture.pool fixture.dlqName+      firstPage <- browseDeadLetters admin fixture.dlqName (DlqPageRequest 2 Nothing) >>= right+      V.map (\item -> item.rawMessage.messageId) firstPage.items `shouldBe` V.fromList [firstId, corruptId]+      firstPage.nextCursor `shouldBe` Just (DlqCursor fixture.dlqName corruptId)+      (firstPage.items V.! 1).decodedEntry `shouldSatisfy` isLeft+      lastPage <- browseDeadLetters admin fixture.dlqName (DlqPageRequest 2 firstPage.nextCursor) >>= right+      V.map (\item -> item.rawMessage.messageId) lastPage.items `shouldBe` V.singleton lastId+      lastPage.nextCursor `shouldBe` Nothing+      nullEntry <- right (V.head lastPage.items).decodedEntry+      nullEntry.originalPayload `shouldBe` Null+      forM_ before $ \msg -> void $ fetchDeadLetter admin fixture.dlqName msg.messageId >>= right+      snapshot fixture.pool fixture.dlqName `shouldReturn` before+      void $ runPgmqSession fixture.pool $ S.deleteMessage (Q.MessageQuery fixture.dlqName firstId)+      browseDeadLetters admin fixture.dlqName (DlqPageRequest 2 firstPage.nextCursor) `shouldReturn` Right lastPage+      exact <- browseDeadLetters admin fixture.dlqName (DlqPageRequest 2 Nothing) >>= right+      V.length exact.items `shouldBe` 2+      exact.nextCursor `shouldBe` Nothing+      fetchDeadLetter admin fixture.dlqName firstId `shouldReturn` Left (EntryNotFound fixture.dlqName firstId)++    it "decodes structured-only, dual-written and legacy history while preserving arbitrary JSON detail" $ \fixture -> do+      admin <- requireAdmin fixture False+      let detail = object ["nested" .= [Null, object ["ok" .= True]]]+          structured = object ["original_message" .= Null, "dead_letter_reason_code" .= ("invalid_payload" :: Text), "dead_letter_reason_detail" .= detail]+          legacy = object ["original_message" .= String "old", "dead_letter_reason" .= ("poison_pill: 原因: retained" :: Text)]+          bare = object ["original_message" .= Null, "dead_letter_reason" .= ("max_retries_exceeded" :: Text)]+      ids <- mapM (seed fixture) [structured, putField "dead_letter_reason" (String "ignored: legacy") structured, legacy, bare]+      entries <- mapM (\messageId -> fetchDeadLetter admin fixture.dlqName messageId >>= right >>= right . (\item -> item.decodedEntry)) ids+      map (\entry -> entry.reasonCode) entries `shouldBe` ["invalid_payload", "invalid_payload", "poison_pill", "max_retries_exceeded"]+      map (\entry -> entry.reasonDetail) entries `shouldBe` [detail, detail, String "原因: retained", Null]+      map (\entry -> entry.reasonRepresentation) entries `shouldBe` [StructuredReason, StructuredReason, LegacyReason, LegacyReason]+      forM_ entries $ \entry -> do+        entry.originalReadCount `shouldBe` Nothing+        entry.originalMessageId `shouldBe` Nothing+        entry.originalEnqueuedAt `shouldBe` Nothing+        entry.dlqReadCount `shouldBe` 0+        entry.sourceResolution `shouldBe` ConfiguredFallback+        entry.resolvedSource `shouldBe` Just fixture.queueName++    it "keeps malformed optional metadata and partial structured reasons inspectable and refuses redrive" $ \fixture -> do+      admin <- requireAdmin fixture True+      let payload = envelope (String "valid body")+          corrupt =+            [ ("original_message", deleteField "original_message" payload),+              ("dead_letter_reason_code", deleteField "dead_letter_reason_code" payload),+              ("dead_letter_reason_detail", deleteField "dead_letter_reason_detail" payload),+              ("dead_letter_reason_code", putField "dead_letter_reason_code" Null payload),+              ("original_queue", putField "original_queue" (Number 42) payload),+              ("original_queue", putField "original_queue" (String "../bad") payload),+              ("read_count", putField "read_count" (String "2") payload),+              ("original_message_id", putField "original_message_id" Null payload),+              ("original_enqueued_at", putField "original_enqueued_at" (String "not-a-time") payload),+              ("last_read_at", putField "last_read_at" (Number 7) payload)+            ]+      forM_ corrupt $ \(field, body) -> do+        messageId <- seed fixture body+        item <- fetchDeadLetter admin fixture.dlqName messageId >>= right+        case item.decodedEntry of+          Right _ -> expectationFailure "corrupt envelope decoded successfully"+          Left err -> do+            err.field `shouldBe` field+            redriveDeadLetter admin fixture.dlqName messageId `shouldReturn` Left (MalformedEnvelope err)+      rows <- snapshot fixture.pool fixture.dlqName+      page <- browseDeadLetters admin fixture.dlqName defaultDlqPageRequest >>= right+      V.length page.items `shouldBe` length corrupt+      snapshot fixture.pool fixture.dlqName `shouldReturn` rows++    it "validates limits, queue-bound cursors and managed queue identity before querying" $ \fixture -> do+      admin <- requireAdmin fixture False+      forM_ [0, -1, 501, maxBound] $ \limit ->+        browseDeadLetters admin fixture.dlqName (DlqPageRequest limit Nothing) `shouldReturn` Left (InvalidPageRequest (InvalidPageLimit limit))+      browseDeadLetters admin fixture.dlqName (DlqPageRequest 50 (Just (DlqCursor fixture.queueName (P.MessageId 1))))+        `shouldReturn` Left (InvalidPageRequest (CursorQueueMismatch fixture.dlqName fixture.queueName))+      browseDeadLetters admin fixture.queueName defaultDlqPageRequest `shouldReturn` Left (UnknownManagedDlq fixture.queueName)+      fetchDeadLetter admin fixture.queueName (P.MessageId 1) `shouldReturn` Left (UnknownManagedDlq fixture.queueName)+      redriveDeadLetter admin fixture.queueName (P.MessageId 1) `shouldReturn` Left RedriveDisabled++    it "does not turn missing tables into an empty page or EntryNotFound" $ \fixture -> do+      admin <- requireAdmin fixture False+      void $ runPgmqSession fixture.pool $ S.dropQueue fixture.dlqName+      browseDeadLetters admin fixture.dlqName defaultDlqPageRequest >>= (`shouldSatisfy` isDatabaseError)+      fetchDeadLetter admin fixture.dlqName (P.MessageId 1) >>= (`shouldSatisfy` isDatabaseError)+      runPgmqSession fixture.pool $ S.createQueue fixture.dlqName++    it "requires valid direct mappings and positive deadlines at handle construction" $ \fixture -> do+      let queue = fixture.dlqName+          source = fixture.queueName+          registration = DlqRegistration queue [source]+          config = defaultDeadLetterAdminConfig [registration]+          rejects cfg expected = case mkDeadLetterAdmin fixture.pool cfg of+            Left err -> err `shouldBe` expected+            Right _ -> expectationFailure "invalid config accepted"+      rejects (config {registrations = [registration, registration]}) (DuplicateManagedQueue queue)+      rejects (config {registrations = [DlqRegistration queue []]}) (EmptySourceQueues queue)+      rejects (config {registrations = [DlqRegistration queue [source, source]]}) (DuplicateSourceQueues queue)+      rejects (config {registrations = [DlqRegistration queue [queue]]}) (SelfMapping queue)+      rejects (config {statementTimeoutMs = 0}) (InvalidStatementTimeout 0)+      rejects (config {lockTimeoutMs = -1}) (InvalidLockTimeout (-1))+      registrationFromAdapterConfig (defaultConfig source) `shouldBe` Left NoDirectDeadLetterTarget+      registrationFromAdapterConfig ((defaultConfig source) {deadLetterConfig = Just (directDeadLetter queue False)}) `shouldBe` Right registration+      let key = either (error . show) id (P.parseRoutingKey "fail.orders")+      registrationFromAdapterConfig ((defaultConfig source) {deadLetterConfig = Just (topicDeadLetter key True)}) `shouldBe` Left NoDirectDeadLetterTarget+      registrationFromAdapterConfig ((defaultConfig source) {deadLetterConfig = Just (directDeadLetter source True)}) `shouldBe` Left (SelfMapping source)++    it "resolves registered shared DLQs by payload and refuses ambiguous or disallowed sources" $ \fixture -> do+      let other = namedQueue "another_source"+          forbidden = namedQueue "unregistered_source"+      admin <-+        right $+          mkDeadLetterAdmin fixture.pool $+            (defaultDeadLetterAdminConfig [DlqRegistration fixture.dlqName [fixture.queueName, other]]) {redriveEnabled = True}+      explicit <- seed fixture (putField "original_queue" (String (P.queueNameToText other)) (envelope Null))+      ambiguous <- seed fixture (envelope Null)+      disallowed <- seed fixture (putField "original_queue" (String (P.queueNameToText forbidden)) (envelope Null))+      item <- fetchDeadLetter admin fixture.dlqName explicit >>= right+      entry <- right item.decodedEntry+      entry.resolvedSource `shouldBe` Just other+      entry.sourceResolution `shouldBe` PayloadSource+      rows <- snapshot fixture.pool fixture.dlqName+      redriveDeadLetter admin fixture.dlqName ambiguous `shouldReturn` Left UnresolvedSource+      redriveDeadLetter admin fixture.dlqName disallowed `shouldReturn` Left (DisallowedSource forbidden)+      snapshot fixture.pool fixture.dlqName `shouldReturn` rows++    it "redrives the original payload once and preserves saved group/trace headers" $ \fixture -> do+      admin <- requireAdmin fixture True+      let body = object ["order" .= (42 :: Int)]+          saved = object ["x-pgmq-group" .= ("customer-a" :: Text), "traceparent" .= ("original" :: Text)]+          wrapper = putField "original_headers" saved (envelope body)+      messageId <- seedWithHeaders fixture wrapper (object ["traceparent" .= ("dlq" :: Text)])+      before <- snapshot fixture.pool fixture.dlqName+      disabled <- requireAdmin fixture False+      redriveDeadLetter disabled fixture.dlqName messageId `shouldReturn` Left RedriveDisabled+      snapshot fixture.pool fixture.dlqName `shouldReturn` before+      result <- redriveDeadLetter admin fixture.dlqName messageId >>= right+      newId <- case result of+        Redriven value -> pure value+        AlreadyAbsent -> expectationFailure "fresh row absent" >> fail "absent"+      snapshot fixture.pool fixture.dlqName `shouldReturn` V.empty+      sourceRows <- snapshot fixture.pool fixture.queueName+      V.length sourceRows `shouldBe` 1+      let restored = V.head sourceRows+      restored.messageId `shouldBe` newId+      restored.body `shouldBe` P.MessageBody body+      restored.headers `shouldBe` Just saved+      restored.readCount `shouldBe` 0+      redriveDeadLetter admin fixture.dlqName messageId `shouldReturn` Right AlreadyAbsent+      -- Discarded confirmed response followed by a fresh retry, not network loss.+      redriveDeadLetter admin fixture.dlqName messageId `shouldReturn` Right AlreadyAbsent+      snapshot fixture.pool fixture.queueName `shouldReturn` sourceRows+      consumed <- runPgmqSession fixture.pool $ S.readMessage (Q.ReadMessage fixture.queueName 30 (Just 1) Nothing)+      V.map (\msg -> msg.body) consumed `shouldBe` V.singleton (P.MessageBody body)++    it "distinguishes missing original headers from explicit null and accepts null payload" $ \fixture -> do+      admin <- requireAdmin fixture True+      let trace = object ["traceparent" .= ("dlq-span" :: Text), "x-pgmq-group" .= ("g" :: Text)]+      explicitNull <- seedWithHeaders fixture (putField "original_headers" Null (envelope Null)) trace+      missing <- seedWithHeaders fixture (envelope (String "metadata-disabled")) trace+      forM_ [explicitNull, missing] $ \messageId -> void $ redriveDeadLetter admin fixture.dlqName messageId >>= right+      rows <- snapshot fixture.pool fixture.queueName+      V.map (\msg -> msg.body) rows `shouldBe` V.fromList [P.MessageBody Null, P.MessageBody (String "metadata-disabled")]+      V.map (\msg -> msg.headers) rows `shouldBe` V.fromList [Just Null, Just trace]++    it "restores arbitrary JSON-valued saved headers supported by PGMQ" $ \fixture -> do+      admin <- requireAdmin fixture True+      let headers = [String "scalar", Array (V.fromList [Null, Bool True]), Number 7, Bool False]+      forM_ headers $ \saved -> do+        messageId <- seed fixture (putField "original_headers" saved (envelope Null))+        redriveDeadLetter admin fixture.dlqName messageId >>= (`shouldSatisfy` isRedriven)+      rows <- snapshot fixture.pool fixture.queueName+      V.map (\msg -> msg.headers) rows `shouldBe` V.fromList (map Just headers)++    it "serializes independent concurrent redrives on one durable row claim" $ \fixture -> do+      admin <- requireAdmin fixture True+      messageId <- seed fixture (envelope (String "race"))+      results <- concurrently (redriveDeadLetter admin fixture.dlqName messageId) (redriveDeadLetter admin fixture.dlqName messageId)+      case results of+        (Right (Redriven _), Right AlreadyAbsent) -> pure ()+        (Right AlreadyAbsent, Right (Redriven _)) -> pure ()+        _ -> expectationFailure (show results)+      rows <- snapshot fixture.pool fixture.queueName+      V.length rows `shouldBe` 1+      snapshot fixture.pool fixture.dlqName `shouldReturn` V.empty++    it "rolls the deletion back on send failure and can retry after the source is recreated" $ \fixture -> do+      admin <- requireAdmin fixture True+      messageId <- seed fixture (envelope (String "rollback"))+      before <- snapshot fixture.pool fixture.dlqName+      void $ runPgmqSession fixture.pool $ S.dropQueue fixture.queueName+      redriveDeadLetter admin fixture.dlqName messageId >>= (`shouldSatisfy` isDatabaseError)+      snapshot fixture.pool fixture.dlqName `shouldReturn` before+      runPgmqSession fixture.pool $ S.createQueue fixture.queueName+      redriveDeadLetter admin fixture.dlqName messageId >>= (`shouldSatisfy` isRedriven)+      rows <- snapshot fixture.pool fixture.queueName+      V.map (\msg -> msg.body) rows `shouldBe` V.singleton (P.MessageBody (String "rollback"))++    it "treats topic fan-out copies as two separately redrivable entries" $ \fixture -> do+      let other = namedQueue "topic_dlq_copy"+          key = either (error . show) id (P.parseRoutingKey "fail.orders")+          pattern' = either (error . show) id (P.parseTopicPattern "fail.#")+      runPgmqSession fixture.pool $ do+        S.createQueue other+        S.bindTopic (Q.BindTopic pattern' fixture.dlqName)+        S.bindTopic (Q.BindTopic pattern' other)+      produceFailure+        fixture+        ( (defaultConfig fixture.queueName)+            { batchSize = 1,+              polling = StandardPolling 0.01,+              deadLetterConfig = Just (topicDeadLetter key True)+            }+        )+        (String "fanout")+      admin <-+        right $+          mkDeadLetterAdmin fixture.pool $+            (defaultDeadLetterAdminConfig [DlqRegistration q [fixture.queueName] | q <- [fixture.dlqName, other]]) {redriveEnabled = True}+      copies <-+        mapM+          ( \q -> do+              page <- browseDeadLetters admin q defaultDlqPageRequest >>= right+              V.length page.items `shouldBe` 1+              pure (q, (V.head page.items).rawMessage.messageId)+          )+          [fixture.dlqName, other]+      ((q1, id1), (q2, id2)) <- case copies of+        [first, second] -> pure (first, second)+        _ -> expectationFailure "expected exactly two topic copies" >> fail "copy count"+      (r1, r2) <- concurrently (redriveDeadLetter admin q1 id1) (redriveDeadLetter admin q2 id2)+      r1 `shouldSatisfy` isRedriven+      r2 `shouldSatisfy` isRedriven+      rows <- snapshot fixture.pool fixture.queueName+      V.map (\msg -> msg.body) rows `shouldBe` V.replicate 2 (P.MessageBody (String "fanout"))+      void $ runPgmqSession fixture.pool $ S.dropQueue other++    it "redrives real metadata-disabled writes through a normal handler with the original payload" $ \fixture -> do+      let config =+            (defaultConfig fixture.queueName)+              { batchSize = 1,+                polling = StandardPolling 0.01,+                deadLetterConfig = Just (directDeadLetter fixture.dlqName False)+              }+      produceFailure fixture config (String "no metadata")+      admin <- requireAdmin fixture True+      page <- browseDeadLetters admin fixture.dlqName defaultDlqPageRequest >>= right+      let item = V.head page.items+      entry <- right item.decodedEntry+      entry.originalReadCount `shouldBe` Nothing+      entry.originalHeaders `shouldBe` Nothing+      entry.sourceResolution `shouldBe` PayloadSource+      redriveDeadLetter admin fixture.dlqName item.rawMessage.messageId >>= (`shouldSatisfy` isRedriven)+      -- Run an ordinary handler against the source and observe its public input.+      received <- newEmptyMVar+      outcome <- runEff $ runErrorNoCallStack @PgmqRuntimeError $ runPgmq fixture.pool $ runTracingNoop $ do+        adapter <- pgmqAdapter (mkPgmqAdapterEnv fixture.pool) config >>= either (error . show) pure+        app <- runApp defaultAppConfig [(ProcessorId "redriven-input", mkProcessor adapter (\message -> liftIO (putMVar received message.envelope.payload) >> pure AckOk))] >>= either (error . show) pure+        actual <- liftIO $ timeout 5000000 (takeMVar received)+        liftIO $ actual `shouldBe` Just (String "no metadata")+        void $ stopAppGracefully defaultShutdownConfig app+      either throwIO pure outcome++    it "includes source identity for automatic dead-letter moves" $ \fixture -> do+      let config =+            (defaultConfig fixture.queueName)+              { maxRetries = 0,+                batchSize = 1,+                polling = StandardPolling 0.01,+                deadLetterConfig = Just (directDeadLetter fixture.dlqName False)+              }+      produceFailure fixture config (String "automatic")+      admin <- requireAdmin fixture False+      page <- browseDeadLetters admin fixture.dlqName defaultDlqPageRequest >>= right+      entry <- right (V.head page.items).decodedEntry+      entry.reasonCode `shouldBe` "max_retries_exceeded"+      entry.resolvedSource `shouldBe` Just fixture.queueName+      entry.originalReadCount `shouldBe` Nothing++    it "returns a full 500-row page with one lookahead row and an exact boundary" $ \fixture -> do+      ids <- mapM (\_ -> seed fixture (envelope Null)) [1 .. 501 :: Int]+      admin <- requireAdmin fixture False+      page <- browseDeadLetters admin fixture.dlqName (DlqPageRequest 500 Nothing) >>= right+      V.length page.items `shouldBe` 500+      page.nextCursor `shouldBe` Just (DlqCursor fixture.dlqName (ids !! 499))+      next <- browseDeadLetters admin fixture.dlqName (DlqPageRequest 500 page.nextCursor) >>= right+      V.map (\item -> item.rawMessage.messageId) next.items `shouldBe` V.singleton (last ids)+      next.nextCursor `shouldBe` Nothing+  administrativePoolSpec++database :: (TestFixture -> IO ()) -> IO ()+database action = withPgmqDb (\pool -> withTestFixture pool action) >>= either (error . show) pure++right :: (Show e) => Either e a -> IO a+right = either (\err -> expectationFailure (show err) >> fail "expected Right") pure++requireAdmin :: TestFixture -> Bool -> IO DeadLetterAdmin+requireAdmin fixture enabled =+  right $+    mkDeadLetterAdmin fixture.pool $+      (defaultDeadLetterAdminConfig [DlqRegistration fixture.dlqName [fixture.queueName]]) {redriveEnabled = enabled}++awaitDlq :: Pool.Pool -> P.QueueName -> IO ()+awaitDlq pool queue = go (500 :: Int)+  where+    go 0 = expectationFailure "handler did not commit its DLQ move"+    go n = do+      rows <- runPgmqSession pool $ S.peekMessages (Q.PeekMessages (P.queueNameToText queue) Nothing 1)+      if V.null rows then threadDelay 10000 >> go (n - 1) else pure ()++envelope :: Value -> Value+envelope body =+  object+    [ "original_message" .= body,+      "dead_letter_reason_code" .= ("orders.validation.failed" :: Text),+      "dead_letter_reason_detail" .= String "detail"+    ]++putField :: Data.Aeson.Key.Key -> Value -> Value -> Value+putField field value (Object fields) = Object (KM.insert field value fields)+putField _ _ _ = error "expected object fixture"++deleteField :: Data.Aeson.Key.Key -> Value -> Value+deleteField field (Object fields) = Object (KM.delete field fields)+deleteField _ _ = error "expected object fixture"++seed :: TestFixture -> Value -> IO P.MessageId+seed fixture body = runPgmqSession fixture.pool $ S.sendMessage (Q.SendMessage fixture.dlqName (P.MessageBody body) Nothing)++seedWithHeaders :: TestFixture -> Value -> Value -> IO P.MessageId+seedWithHeaders fixture body headers =+  runPgmqSession fixture.pool $+    S.sendMessageWithHeaders (Q.SendMessageWithHeaders fixture.dlqName (P.MessageBody body) (P.MessageHeaders headers) Nothing)++snapshot :: Pool.Pool -> P.QueueName -> IO (V.Vector P.Message)+snapshot pool q = runPgmqSession pool $ S.peekMessages (Q.PeekMessages (P.queueNameToText q) Nothing 10000)++namedQueue :: Text -> P.QueueName+namedQueue = either (error . show) id . P.parseQueueName++isDatabaseError :: Either DlqError a -> Bool+isDatabaseError (Left (DatabaseError _)) = True+isDatabaseError _ = False++isRedriven :: Either DlqError RedriveOutcome -> Bool+isRedriven (Right (Redriven _)) = True+isRedriven _ = False++produceFailure :: TestFixture -> PgmqAdapterConfig -> Value -> IO ()+produceFailure fixture config body = do+  void $ runPgmqSession fixture.pool $ S.sendMessage (Q.SendMessage fixture.queueName (P.MessageBody body) Nothing)+  outcome <- runEff $ runErrorNoCallStack @PgmqRuntimeError $ runPgmq fixture.pool $ runTracingNoop $ do+    adapter <- pgmqAdapter (mkPgmqAdapterEnv fixture.pool) config >>= either (error . show) pure+    app <- runApp defaultAppConfig [(ProcessorId "source-identity", mkProcessor adapter (\_ -> pure (AckDeadLetter (InvalidPayload "failed"))))] >>= either (error . show) pure+    liftIO $ awaitDlq fixture.pool fixture.dlqName+    void $ stopAppGracefully defaultShutdownConfig app+  either throwIO pure outcome++administrativePoolSpec :: Spec+administrativePoolSpec = do+  it "scopes statement and lock deadlines to each transaction on a borrowed single-connection pool" $+    withPools $ \fixture adminPool -> do+      admin <- right $ mkDeadLetterAdmin adminPool (defaultDeadLetterAdminConfig [DlqRegistration fixture.dlqName [fixture.queueName]])+      before <- poolSettings adminPool+      messageId <- seed fixture (envelope Null)+      void $ browseDeadLetters admin fixture.dlqName defaultDlqPageRequest >>= right+      void $ fetchDeadLetter admin fixture.dlqName messageId >>= right+      poolSettings adminPool `shouldReturn` before+      rows <- snapshot fixture.pool fixture.dlqName+      let config =+            (defaultDeadLetterAdminConfig [DlqRegistration fixture.dlqName [fixture.queueName]])+              { redriveEnabled = True,+                statementTimeoutMs = 1000,+                lockTimeoutMs = 20+              }+      bounded <- right $ mkDeadLetterAdmin adminPool config+      withLockedQueue fixture.pool fixture.dlqName $ do+        result <- timeout 1000000 (redriveDeadLetter bounded fixture.dlqName messageId)+        result `shouldSatisfy` maybe False isDatabaseError+      poolSettings adminPool `shouldReturn` before+      snapshot fixture.pool fixture.dlqName `shouldReturn` rows+      -- A shorter statement deadline also interrupts read-side lock waits.+      shortStatement <- right $ mkDeadLetterAdmin adminPool (config {statementTimeoutMs = 20, lockTimeoutMs = 1000})+      withLockedQueue fixture.pool fixture.dlqName $ do+        result <- timeout 1000000 (browseDeadLetters shortStatement fixture.dlqName defaultDlqPageRequest)+        result `shouldSatisfy` maybe False isDatabaseError+      poolSettings adminPool `shouldReturn` before+      redriveDeadLetter bounded fixture.dlqName messageId >>= (`shouldSatisfy` isRedriven)+      poolSettings adminPool `shouldReturn` before++  it "lets cancellation escape without a successful absence or a lost DLQ row" $+    withPools $ \fixture adminPool -> do+      admin <-+        right $+          mkDeadLetterAdmin adminPool $+            (defaultDeadLetterAdminConfig [DlqRegistration fixture.dlqName [fixture.queueName]]) {redriveEnabled = True}+      messageId <- seed fixture (envelope (String "cancelled"))+      before <- snapshot fixture.pool fixture.dlqName+      withLockedQueue fixture.pool fixture.dlqName $+        withAsync (redriveDeadLetter admin fixture.dlqName messageId) $ \request -> do+          awaitBlockedRedrive fixture.pool fixture.dlqName+          cancel request+          waitCatch request >>= (`shouldSatisfy` isLeft)+      snapshot fixture.pool fixture.dlqName `shouldReturn` before+      redriveDeadLetter admin fixture.dlqName messageId >>= (`shouldSatisfy` isRedriven)++  it "checks disabled mutations and invalid requests even with an unusable database pool" $ do+    let settings = Settings.connectionString "host=/tmp/plan9-nonexistent-postgres connect_timeout=1"+        config =+          PoolConfig.settings+            [ PoolConfig.size 1,+              PoolConfig.acquisitionTimeout (secondsToDiffTime 1),+              PoolConfig.staticConnectionSettings settings+            ]+        source = namedQueue "source"+        dlq = namedQueue "dlq"+    bracket (Pool.acquire config) Pool.release $ \pool -> do+      admin <- right $ mkDeadLetterAdmin pool (defaultDeadLetterAdminConfig [DlqRegistration dlq [source]])+      redriveDeadLetter admin source (P.MessageId 1) `shouldReturn` Left RedriveDisabled+      browseDeadLetters admin dlq (DlqPageRequest 0 Nothing) `shouldReturn` Left (InvalidPageRequest (InvalidPageLimit 0))+      fetchDeadLetter admin source (P.MessageId 1) `shouldReturn` Left (UnknownManagedDlq source)+      fetchDeadLetter admin dlq (P.MessageId 1) >>= (`shouldSatisfy` isDatabaseError)++withPools :: (TestFixture -> Pool.Pool -> IO ()) -> IO ()+withPools action = withPgmqDbPools 4 1 (\workers admin -> withTestFixture workers (\fixture -> action fixture admin)) >>= either (error . show) pure++poolSettings :: Pool.Pool -> IO (Text, Text)+poolSettings pool =+  runPgmqSession pool $+    Session.statement () $+      Statement.unpreparable+        "select current_setting('statement_timeout'), current_setting('lock_timeout')"+        E.noParams+        (D.singleRow ((,) <$> D.column (D.nonNullable D.text) <*> D.column (D.nonNullable D.text)))++withLockedQueue :: Pool.Pool -> P.QueueName -> IO a -> IO a+withLockedQueue pool q action = do+  ready <- newEmptyMVar+  release <- newEmptyMVar+  let sql statement = Session.statement () (Statement.unpreparable statement E.noParams D.noResult)+      hold = runPgmqSession pool $ do+        sql "begin"+        sql ("lock table pgmq." <> Inspection.quoteIdentifier ("q_" <> P.queueNameToText q) <> " in access exclusive mode")+        liftIO (putMVar ready ())+        liftIO (takeMVar release)+        sql "rollback"+  bracket (async hold) (\worker -> putMVar release () >> void (waitCatch worker)) $ \_ -> do+    acquired <- timeout 5000000 (takeMVar ready)+    acquired `shouldBe` Just ()+    action++-- Wait for the request to be inside its transaction, rather than merely+-- cancelling a thread which might still be acquiring its first connection.+awaitBlockedRedrive :: Pool.Pool -> P.QueueName -> IO ()+awaitBlockedRedrive pool q = go (200 :: Int)+  where+    patternText = "%from pgmq." <> Inspection.quoteIdentifier ("q_" <> P.queueNameToText q) <> "%for update"+    waiting =+      Session.statement patternText $+        Statement.unpreparable+          "select exists(select 1 from pg_stat_activity where wait_event_type = 'Lock' and query like $1)"+          (E.param (E.nonNullable E.text))+          (D.singleRow (D.column (D.nonNullable D.bool)))+    go 0 = expectationFailure "redrive did not reach its row-lock wait"+    go n = do+      blocked <- runPgmqSession pool waiting+      if blocked then pure () else threadDelay 10000 >> go (n - 1)

@@ -5,6 +5,7 @@ module TmpPostgres   ( -- * Test Execution     withPgmqDb,+    withPgmqDbPools,     withRestartablePgmqDb,     withTestFixture, @@ -20,6 +21,7 @@ import Control.Monad ((>=>)) import Data.IORef (newIORef, readIORef, writeIORef) import Data.List.NonEmpty (NonEmpty ((:|)))+import Data.Monoid qualified as EphemeralMonoid import Data.Text (Text) import Data.Text qualified as Text import Data.Time (secondsToDiffTime)@@ -35,6 +37,8 @@ import Pgmq.Hasql.Sessions qualified as Pgmq import Pgmq.Migration qualified as Migration import Pgmq.Types (QueueName, parseQueueName)+import System.Directory qualified as EphemeralDirectory+import System.Posix.User qualified as EphemeralUser import System.Random (randomIO)  -- | Test fixture containing pool and queue names for a test@@ -49,7 +53,7 @@ -- This creates an ephemeral PostgreSQL instance, installs the pgmq schema, -- and then runs the provided action with a connection pool. withPgmqDb :: (Pool.Pool -> IO a) -> IO (Either StartError a)-withPgmqDb action = Pg.with $ \db -> do+withPgmqDb action = withEphemeralPg $ \db -> do   let connSettings = Pg.connectionSettings db    -- Install pgmq schema@@ -61,13 +65,21 @@     Pool.release     action +-- | Independent worker and administrative pools against the same disposable DB.+withPgmqDbPools :: Int -> Int -> (Pool.Pool -> Pool.Pool -> IO a) -> IO (Either StartError a)+withPgmqDbPools workerSize adminSize action = withEphemeralPg $ \db -> do+  let settings = Pg.connectionSettings db+  installPgmqSchema settings+  bracket (createPoolWithSize workerSize settings) Pool.release $ \workerPool ->+    bracket (createPoolWithSize adminSize settings) Pool.release (action workerPool)+ -- | Run against an ephemeral database that the test may restart in place. -- The restart preserves the data directory, releases stale pooled connections, -- and lets the next pool use establish a fresh connection to the same socket. withRestartablePgmqDb ::   (Pool.Pool -> IO (Either StartError ()) -> IO a) ->   IO (Either StartError a)-withRestartablePgmqDb action = Pg.with $ \initialDb ->+withRestartablePgmqDb action = withEphemeralPg $ \initialDb ->   bracket     (newIORef initialDb)     (readIORef >=> Pg.stop)@@ -151,13 +163,26 @@  -- | Create a connection pool from connection settings. createPool :: Settings.Settings -> IO Pool.Pool-createPool connSettings = do+createPool = createPoolWithSize 10++createPoolWithSize :: Int -> Settings.Settings -> IO Pool.Pool+createPoolWithSize poolSize connSettings = do   let poolConfig =         PoolConfig.settings-          [ PoolConfig.size 10,+          [ PoolConfig.size poolSize,             PoolConfig.acquisitionTimeout (secondsToDiffTime 10),             PoolConfig.agingTimeout (secondsToDiffTime 3600),             PoolConfig.idlenessTimeout (secondsToDiffTime 60),             PoolConfig.staticConnectionSettings connSettings           ]   Pool.acquire poolConfig++-- | Stable per-user root lets the next invocation reap abandoned clusters.+-- See mori://shinzui/ephemeral-pg/docs/guides (temporary-roots-and-stale-cleanup.md; artifact URI pending).+withEphemeralPg :: (Pg.Database -> IO a) -> IO (Either Pg.StartError a)+withEphemeralPg action = do+  uid <- EphemeralUser.getEffectiveUserID+  let root = "/tmp/ephpg-shibuya-pgmq-adapter-" <> show uid+  EphemeralDirectory.createDirectoryIfMissing True root+  let config = Pg.defaultConfig {Pg.temporaryRoot = EphemeralMonoid.Last (Just root)}+  Pg.withConfig config action