pgmq-effectful 0.6.1.1 → 0.7.0.0

raw patch · 13 files changed

+879/−56 lines, 13 filesdep ~ephemeral-pgdep ~pgmq-coredep ~pgmq-effectfulPVP ok

version bump matches the API change (PVP)

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

API changes (from Hackage documentation)

Files

• CHANGELOG.md +57/−0
• pgmq-effectful.cabal +10/−8
• src/Pgmq/Effectful.hs +24/−1
• src/Pgmq/Effectful/Effect.hs +54/−1
• src/Pgmq/Effectful/Interpreter.hs +87/−13
• src/Pgmq/Effectful/Interpreter/Traced.hs +28/−1
• test/ClassificationSpec.hs +118/−16
• test/DisconnectSpec.hs +324/−0
• test/EphemeralDb.hs +42/−11
• test/Main.hs +7/−3
• test/RootPermissionsSpec.hs +63/−0
• test/TracedInterpreterSpec.hs +51/−1
• test/UmbrellaExportsSpec.hs +14/−1

@@ -1,5 +1,62 @@ # Revision history for pgmq-effectful +## 0.7.0.0 -- 2026-10-26++### Bug Fixes++- `isTransient` now classifies a connection lost while waiting for a reply as transient+  (BUG-1). hasql reports such a loss through statement constructors rather than+  `ConnectionSessionError`, in one of two shapes, and both used to be permanent. The first+  is a `ServerError` with an empty SQLSTATE: libpq's own connection-loss result, which+  carries no fields. Every `ErrorResponse` PostgreSQL sends carries a SQLSTATE, so an empty+  code is always client-side. It appears after SIGKILL of the backend, an immediate server+  shutdown, or a TCP reset. The second is `UnexpectedRowCountStatementError 1 1 1`, hasql's+  encoding of a stray second result, which appears when the server sends a FATAL such as+  `57P01` and libpq then appends its own connection-loss result. Only that exact value is+  transient; a genuine row-count mismatch never reports an actual count inside its bounds,+  so `1 1 0`, `1 1 2`, and every other count stay permanent.+- `ScriptSessionError` now follows the same SQLSTATE rule as statement errors. Previously+  every script error was permanent, including `57P01`.+- Every other server SQLSTATE, decode, missing-type, and driver-error policy is unchanged.+  After either disconnect shape the pool keeps the dead connection, so a retry loop sees one+  more transient `ConnectionSessionError` (which evicts it) before a fresh connection+  succeeds.++### New Features++- `isAmbiguousReply :: PgmqRuntimeError -> Bool`, exported from `Pgmq.Effectful` and+  `Pgmq.Effectful.Interpreter`, answers `True` exactly when a statement was delivered but+  no server verdict came back: an empty-SQLSTATE server error in a statement or script, or+  `UnexpectedRowCountStatementError 1 1 1`. Every ambiguous reply is also transient. Retrying+  a lost send reply blindly can duplicate a message that already committed, so gate the+  retry of non-idempotent operations on `isAmbiguousReply` and reconcile first. See the+  bounded, sticky loop in the README and `docs/design/017-transient-error-classification.md`.+- Non-destructive inspection on the `Pgmq` effect (IR-1): `peekMessages`,+  `peekArchivedMessages`, `lookupMessage`, `lookupArchivedMessage`, and+  `queueMetricsUnvalidated`, re-exported from `Pgmq.Effectful` with `PeekMessages`,+  `LookupMessage`, and `ArchivedMessage`. Both interpreters run them; the traced one labels+  the four hand-written reads `pgmq.peek`, `pgmq.peek_archive`, `pgmq.lookup_message`, and+  `pgmq.lookup_archived_message` (`Internal` spans; lookups carry `messaging.message.id`)+  and keeps `pgmq.metrics` for the lenient metrics read. See+  `docs/design/019-non-destructive-inspection-reads.md`.++### Breaking Changes++- The `Pgmq` effect gained five constructors (`PeekMessages`, `PeekArchivedMessages`,+  `LookupMessage`, `LookupArchivedMessage`, `QueueMetricsUnvalidated`). An interpreter that+  matches every constructor, such as a mock in a consumer's test suite, must add five cases.++### Other Changes++Test infrastructure: create and validate owner-only (`0700`) ephemeral roots, tighten+existing owned directories, reject unsafe paths and preserve lifecycle registry contents.++- Clarify that enableNotifyInsert does not deliver on a partitioned queue's documented parent channel.++Inspection Haddocks now state the moving-view cursor contract and the need for+finite database deadlines and inspection capacity separate from workers; a row limit+does not bound scan work or payload bytes.+ ## 0.6.1.1 -- 2026-09-25  Coordinated family version bump; the effect layer is unchanged from 0.6.1.0.

@@ -1,6 +1,6 @@ cabal-version: 3.4 name: pgmq-effectful-version: 0.6.1.1+version: 0.7.0.0 synopsis: Effectful effects for PGMQ (PostgreSQL Message Queue) description:   Effectful effects and interpreters for pgmq-hs, a Haskell client@@ -64,8 +64,8 @@     hs-opentelemetry-api >=1.0 && <2,     hs-opentelemetry-semantic-conventions >=1.40 && <2,     http-types >=0.12 && <0.13,-    pgmq-core >=0.6 && <0.7,-    pgmq-hasql >=0.6 && <0.7,+    pgmq-core >=0.7 && <0.8,+    pgmq-hasql >=0.7 && <0.8,     text >=2.0 && <2.2,     unliftio >=0.2 && <0.3,     unordered-containers >=0.2 && <0.3,@@ -87,8 +87,10 @@    other-modules:     ClassificationSpec+    DisconnectSpec     EphemeralDb     PlainInterpreterSpec+    RootPermissionsSpec     TracedInterpreterSpec     UmbrellaExportsSpec @@ -105,7 +107,7 @@     base >=4.18 && <5,     directory,     effectful-core ^>=2.6 || ^>=2.7,-    ephemeral-pg >=0.3.1.0,+    ephemeral-pg >=0.3.1 && <0.4,     hasql ^>=1.10,     hasql-pool ^>=1.4,     hs-opentelemetry-api >=1.0 && <2,@@ -114,10 +116,10 @@     hs-opentelemetry-sdk >=1.0 && <2,     hs-opentelemetry-semantic-conventions >=1.40 && <2,     pg-migrate,-    pgmq-core >=0.6 && <0.7,-    pgmq-effectful >=0.6 && <0.7,-    pgmq-hasql >=0.6 && <0.7,-    pgmq-migration >=0.6 && <0.7,+    pgmq-core >=0.7 && <0.8,+    pgmq-effectful >=0.7 && <0.8,+    pgmq-hasql >=0.7 && <0.8,+    pgmq-migration >=0.7 && <0.8,     random ^>=1.2,     tasty ^>=1.5,     tasty-hunit ^>=0.10,

@@ -15,6 +15,7 @@     PgmqRuntimeError (..),     fromUsageError,     isTransient,+    isAmbiguousReply,      -- ** Deprecated Error Types     PgmqError (..),@@ -106,11 +107,24 @@     queueMetrics,     allQueueMetrics, +    -- * Non-destructive Inspection++    -- | Reads that observe a queue or its archive without leasing anything.+    -- See @docs/design/019-non-destructive-inspection-reads.md@.+    peekMessages,+    peekArchivedMessages,+    lookupMessage,+    lookupArchivedMessage,+    queueMetricsUnvalidated,+    PeekMessages (..),+    LookupMessage (..),+     -- * Types     MessageBody (..),     MessageHeaders (..),     MessageId (..),     Message (..),+    ArchivedMessage (..),     Queue (..),     QueueName,     SendMessage (..),@@ -192,8 +206,13 @@     listQueues,     listTopicBindings,     listTopicBindingsForQueue,+    lookupArchivedMessage,+    lookupMessage,+    peekArchivedMessages,+    peekMessages,     pop,     queueMetrics,+    queueMetricsUnvalidated,     readGrouped,     readGroupedHead,     readGroupedHeadWithPoll,@@ -218,6 +237,7 @@   ( PgmqError (..),     PgmqRuntimeError (..),     fromUsageError,+    isAmbiguousReply,     isTransient,     runPgmq,   )@@ -252,7 +272,9 @@     BindTopic (..),     CreatePartitionedQueue (..),     EnableNotifyInsert (..),+    LookupMessage (..),     MessageQuery (..),+    PeekMessages (..),     PopMessage (..),     QueueMetrics (..),     ReadGrouped (..),@@ -270,7 +292,8 @@     VisibilityTimeoutQuery (..),   ) import Pgmq.Types-  ( Message (..),+  ( ArchivedMessage (..),+    Message (..),     MessageBody (..),     MessageHeaders (..),     MessageId (..),

@@ -85,6 +85,13 @@     listFifoIndexQueueNames,     queueMetrics,     allQueueMetrics,++    -- * Non-destructive Inspection+    peekMessages,+    peekArchivedMessages,+    lookupMessage,+    lookupArchivedMessage,+    queueMetricsUnvalidated,   ) where @@ -108,7 +115,9 @@     BindTopic,     CreatePartitionedQueue,     EnableNotifyInsert,+    LookupMessage,     MessageQuery,+    PeekMessages,     PopMessage,     QueueMetrics,     ReadGrouped,@@ -127,7 +136,8 @@     VisibilityTimeoutQuery,   ) import Pgmq.Types-  ( Message,+  ( ArchivedMessage,+    Message,     MessageId,     NotifyInsertThrottle,     Queue,@@ -208,6 +218,12 @@   ListFifoIndexQueueNames :: Pgmq m [Text]   QueueMetrics :: QueueName -> Pgmq m QueueMetrics   AllQueueMetrics :: Pgmq m [QueueMetrics]+  -- Non-destructive inspection (no upstream function; design note 019)+  PeekMessages :: PeekMessages -> Pgmq m (Vector Message)+  PeekArchivedMessages :: PeekMessages -> Pgmq m (Vector ArchivedMessage)+  LookupMessage :: LookupMessage -> Pgmq m (Maybe Message)+  LookupArchivedMessage :: LookupMessage -> Pgmq m (Maybe ArchivedMessage)+  QueueMetricsUnvalidated :: Text -> Pgmq m QueueMetrics  type instance DispatchOf Pgmq = 'Dynamic @@ -234,6 +250,11 @@ detachArchive :: (Pgmq :> es) => QueueName -> Eff es () detachArchive = send . DetachArchive +-- | Enable best-effort insert notifications for an ordinary queue.+--+-- Normal partitioned queues receive no insert notifications on their documented+-- parent channel. See @docs/design/015-notification-delivery-contract.md@.+-- The reconciler reports partitioned notification declarations as unsupported. enableNotifyInsert :: (Pgmq :> es) => EnableNotifyInsert -> Eff es () enableNotifyInsert = send . EnableNotifyInsert @@ -432,3 +453,35 @@  allQueueMetrics :: (Pgmq :> es) => Eff es [QueueMetrics] allQueueMetrics = send AllQueueMetrics++-- Non-destructive Inspection++-- | A keyset page of a queue table without leasing anything: @vt@ and+-- @read_ct@ are untouched. Accepts any server-accepted name; pass+-- 'Pgmq.Types.queueNameToText' for a validated one. Pages by exclusive+-- @msg_id@ cursor, never @OFFSET@. A missing queue is a @PgmqSessionError@+-- carrying SQLSTATE @42P01@. Each call has its own snapshot: refresh from the+-- first page for lower IDs committed or archived after the cursor advances.+-- A row limit does not bound scan work or payload bytes. Use finite database+-- deadlines and inspection capacity separate from workers.+peekMessages :: (Pgmq :> es) => PeekMessages -> Eff es (Vector Message)+peekMessages = send . PeekMessages++-- | A keyset page of an archive table, each row with its archival timestamp.+-- Same name, paging, and error rules as 'peekMessages'.+peekArchivedMessages :: (Pgmq :> es) => PeekMessages -> Eff es (Vector ArchivedMessage)+peekArchivedMessages = send . PeekArchivedMessages++-- | One message by id without leasing it, or 'Nothing' when it is not in the+-- queue table (deleted, archived, popped, or never sent).+lookupMessage :: (Pgmq :> es) => LookupMessage -> Eff es (Maybe Message)+lookupMessage = send . LookupMessage++-- | One archived message by id, or 'Nothing' when it is not in the archive.+lookupArchivedMessage :: (Pgmq :> es) => LookupMessage -> Eff es (Maybe ArchivedMessage)+lookupArchivedMessage = send . LookupArchivedMessage++-- | 'queueMetrics' for a queue named by plain text, including names+-- 'Pgmq.Types.parseQueueName' rejects.+queueMetricsUnvalidated :: (Pgmq :> es) => Text -> Eff es QueueMetrics+queueMetricsUnvalidated = send . QueueMetricsUnvalidated

@@ -6,6 +6,7 @@     PgmqRuntimeError (..),     fromUsageError,     isTransient,+    isAmbiguousReply,      -- * Legacy Error Types (deprecated; will be removed in 0.3.0)     PgmqError (..),@@ -56,14 +57,23 @@ -- -- Returns 'True' for acquisition timeouts, networking connection errors, -- uncategorized libpq connection errors, session-level connection drops,--- and server-reported statement errors whose SQLSTATE names a transient--- condition: @40001@ (serialization_failure), @40P01@ (deadlock_detected),--- @55P03@ (lock_not_available), @57P01@ (admin_shutdown), @57P02@--- (crash_shutdown), @57P03@ (cannot_connect_now), and class @53@--- (insufficient resources). All other errors (authentication failure,--- compatibility mismatches, missing types, other statement errors,--- decode\/row-count mismatches, driver bugs) are treated as permanent.+-- server-reported statement or script errors whose SQLSTATE names a+-- transient condition: @40001@ (serialization_failure), @40P01@+-- (deadlock_detected), @55P03@ (lock_not_available), @57P01@+-- (admin_shutdown), @57P02@ (crash_shutdown), @57P03@ (cannot_connect_now),+-- and class @53@ (insufficient resources); server errors in statements or+-- scripts with no SQLSTATE (client-side connection faults, see+-- 'isAmbiguousReply'); and the driver's stray-result shape+-- @UnexpectedRowCountStatementError 1 1 1@. All other errors+-- (authentication failure, compatibility mismatches, missing types, other+-- server errors, genuine decode and row-count mismatches, driver bugs) are+-- treated as permanent. --+-- A connection lost while waiting for a reply leaves the dead connection in+-- the pool, so the next attempt typically fails once more with a transient+-- 'HasqlErrors.ConnectionSessionError' (which evicts it) before a fresh+-- connection succeeds.+-- -- Note: 'HasqlErrors.OtherConnectionError' is classed as transient here -- despite hasql's documentation calling it \"not transient by default\", -- because it is the catch-all for unrecognized libpq errors and classing@@ -79,20 +89,78 @@   PgmqSessionError e -> case e of     HasqlErrors.ConnectionSessionError _ -> True     HasqlErrors.StatementSessionError _ _ _ _ _ statementError ->-      case statementError of-        HasqlErrors.ServerStatementError (HasqlErrors.ServerError code _ _ _ _) ->-          isTransientSqlState code-        _ -> False-    HasqlErrors.ScriptSessionError {} -> False+      isLostReplyStatementError statementError || isTransientStatementError statementError+    HasqlErrors.ScriptSessionError _ serverError ->+      isLostReplyServerError serverError || isTransientServerError serverError     HasqlErrors.MissingTypesSessionError _ -> False     HasqlErrors.DriverSessionError _ -> False +-- | Was the statement delivered but its reply lost, so that the caller cannot+-- tell whether PostgreSQL applied it?+--+-- 'True' for the two values hasql 1.10 produces when the connection dies while+-- it is waiting for a reply: a server error with no SQLSTATE (libpq's own+-- connection-loss result, which carries no fields) inside a statement or script+-- error, and @UnexpectedRowCountStatementError 1 1 1@ (a stray second result;+-- hasql discards the first, so the value cannot show whether that was a server+-- fatal error or a completed command). 'False' for everything else: a+-- 'HasqlErrors.ConnectionSessionError' means the send failed and nothing reached+-- the server, and every other error carries the server's verdict or the+-- driver's own decode verdict.+--+-- Every ambiguous reply is transient, so @isAmbiguousReply err@ implies+-- @isTransient err@. Gate the retry of non-idempotent operations on it: an+-- ambiguous reply to a send may have committed, so check the durable message+-- keys before sending again.+isAmbiguousReply :: PgmqRuntimeError -> Bool+isAmbiguousReply = \case+  PgmqSessionError (HasqlErrors.StatementSessionError _ _ _ _ _ statementError) ->+    isLostReplyStatementError statementError+  PgmqSessionError (HasqlErrors.ScriptSessionError _ serverError) ->+    isLostReplyServerError serverError+  _ -> False++-- | The statement-level shapes of a lost reply (see 'isAmbiguousReply').+--+-- hasql 1.10 reports a stray second result as @UnexpectedRowCountStatementError 1 1 1@+-- (@Hasql.Engine.Errors.fromRecvError@, @TooManyResultsError@ branch). A genuine+-- row-count mismatch can never carry an actual count inside its own bounds, so+-- this exact value is reserved for the driver's stray-result case, which for the+-- single-command roundtrips this library issues only happens when libpq appends+-- its own connection-loss result after a server fatal error (for example+-- @57P01@ from @pg_terminate_backend@). Every other actual count is a real+-- decoder mismatch and stays permanent.+isLostReplyStatementError :: HasqlErrors.StatementError -> Bool+isLostReplyStatementError = \case+  HasqlErrors.ServerStatementError serverError -> isLostReplyServerError serverError+  HasqlErrors.UnexpectedRowCountStatementError 1 1 1 -> True+  _ -> False++-- | Every @ErrorResponse@ PostgreSQL sends carries a SQLSTATE, so an empty code+-- means libpq manufactured the error on the client — \"server closed the+-- connection unexpectedly\" and its relatives — which hasql 1.10 surfaces with+-- every field empty.+isLostReplyServerError :: HasqlErrors.ServerError -> Bool+isLostReplyServerError (HasqlErrors.ServerError code _ _ _ _) = T.null code++-- | Server-reported statement errors worth retrying: those whose SQLSTATE is in+-- the transient whitelist.+isTransientStatementError :: HasqlErrors.StatementError -> Bool+isTransientStatementError = \case+  HasqlErrors.ServerStatementError serverError -> isTransientServerError serverError+  _ -> False++-- | Server errors (from a statement or a script) whose SQLSTATE is in the+-- transient whitelist.+isTransientServerError :: HasqlErrors.ServerError -> Bool+isTransientServerError (HasqlErrors.ServerError code _ _ _ _) = isTransientSqlState code+ -- | SQLSTATEs that indicate a transient, retry-worthy condition: @40001@ -- serialization_failure, @40P01@ deadlock_detected, @55P03@ -- lock_not_available, @57P01@ admin_shutdown, @57P02@ crash_shutdown, -- @57P03@ cannot_connect_now, and class @53@ (insufficient resources — -- 53000\/53100\/53200\/53300\/53400). These arrive as server errors inside--- 'HasqlErrors.StatementSessionError' and are precisely the errors retries+-- 'HasqlErrors.StatementSessionError' or 'HasqlErrors.ScriptSessionError' and are precisely the errors retries -- exist for. Everything else reported by the server is permanent for retry -- purposes. isTransientSqlState :: Text -> Bool@@ -179,6 +247,12 @@   ListFifoIndexQueueNames -> runSession pool Sessions.listFifoIndexQueueNames   QueueMetrics q -> runSession pool $ Sessions.queueMetrics q   AllQueueMetrics -> runSession pool Sessions.allQueueMetrics+  -- Non-destructive inspection+  PeekMessages args -> runSession pool $ Sessions.peekMessages args+  PeekArchivedMessages args -> runSession pool $ Sessions.peekArchivedMessages args+  LookupMessage args -> runSession pool $ Sessions.lookupMessage args+  LookupArchivedMessage args -> runSession pool $ Sessions.lookupArchivedMessage args+  QueueMetricsUnvalidated q -> runSession pool $ Sessions.queueMetricsUnvalidated q  -- Internal helper runSession ::

@@ -41,6 +41,8 @@ -- * Messaging spans: @"publish my-queue"@, @"receive my-queue"@. -- * Lifecycle and observability spans: @"pgmq.archive my-queue"@, --   @"pgmq.set_vt my-queue"@, @"pgmq.list_queues"@ (no destination).+-- * Inspection spans: @"pgmq.peek my-queue"@, @"pgmq.lookup_message my-queue"@+--   (this library's own labels; no pgmq function backs them). -- -- == Span Kinds --@@ -356,6 +358,25 @@   AllQueueMetrics ->     withTracedOp config pool (defaultOpInfo "pgmq.metrics_all" OTel.Internal) $       Sessions.allQueueMetrics+  -- Non-destructive inspection. No pgmq function backs the four reads (they+  -- are hand-written SELECTs, see design note 019), so, like+  -- pgmq.list_fifo_indexes, they carry this library's own labels. The lenient+  -- metrics read does call pgmq.metrics and says so.+  PeekMessages args@(Types.PeekMessages qn _ _) ->+    withTracedOp config pool (queueOpText "pgmq.peek" OTel.Internal qn) $+      Sessions.peekMessages args+  PeekArchivedMessages args@(Types.PeekMessages qn _ _) ->+    withTracedOp config pool (queueOpText "pgmq.peek_archive" OTel.Internal qn) $+      Sessions.peekArchivedMessages args+  LookupMessage args@(Types.LookupMessage qn msgId) ->+    withTracedOp config pool ((queueOpText "pgmq.lookup_message" OTel.Internal qn) {opMessageId = Just msgId}) $+      Sessions.lookupMessage args+  LookupArchivedMessage args@(Types.LookupMessage qn msgId) ->+    withTracedOp config pool ((queueOpText "pgmq.lookup_archived_message" OTel.Internal qn) {opMessageId = Just msgId}) $+      Sessions.lookupArchivedMessage args+  QueueMetricsUnvalidated q ->+    withTracedOp config pool (queueOpText "pgmq.metrics" OTel.Internal q) $+      Sessions.queueMetricsUnvalidated q  -- --------------------------------------------------------------------- -- OpInfo constructors@@ -363,7 +384,13 @@  -- | 'OpInfo' for a non-messaging operation scoped to a single queue. queueOp :: Text -> OTel.SpanKind -> QueueName -> OpInfo-queueOp fn kind qn = (defaultOpInfo fn kind) {opDestination = Just (queueNameToText qn)}+queueOp fn kind qn = queueOpText fn kind (queueNameToText qn)++-- | 'OpInfo' for a non-messaging operation scoped to a queue named by plain+-- text: the lenient inspection path, which accepts names 'Pgmq.Types.parseQueueName'+-- rejects. The destination attribute carries the name exactly as given.+queueOpText :: Text -> OTel.SpanKind -> Text -> OpInfo+queueOpText fn kind name = (defaultOpInfo fn kind) {opDestination = Just name}  -- | 'OpInfo' for a @publish@ (send) on a queue. publishOp :: Text -> QueueName -> OpInfo

@@ -7,6 +7,7 @@ import Hasql.Errors qualified as HasqlErrors import Pgmq.Effectful   ( PgmqRuntimeError (..),+    isAmbiguousReply,     isTransient,   ) import Test.Tasty (TestTree, testGroup)@@ -29,6 +30,69 @@         (HasqlErrors.ServerStatementError (HasqlErrors.ServerError code "boom" Nothing Nothing Nothing))     ) +-- | A server error reported for a 'HasqlErrors.ScriptSessionError' with the+-- given SQLSTATE.+scriptError :: Text -> PgmqRuntimeError+scriptError code =+  PgmqSessionError+    ( HasqlErrors.ScriptSessionError+        "checkpoint"+        (HasqlErrors.ServerError code "boom" Nothing Nothing Nothing)+    )++-- | The value hasql 1.10 surfaces when libpq's own connection-loss result is+-- decoded: a server error with every field empty.+lostReplyStatementError :: PgmqRuntimeError+lostReplyStatementError =+  PgmqSessionError+    ( HasqlErrors.StatementSessionError+        1+        0+        "select 1"+        []+        True+        (HasqlErrors.ServerStatementError (HasqlErrors.ServerError "" "" Nothing Nothing Nothing))+    )++-- | A row-count statement error with the given bounds and actual count.+rowCountError :: Int -> Int -> Int -> PgmqRuntimeError+rowCountError lo hi actual =+  PgmqSessionError+    ( HasqlErrors.StatementSessionError+        1+        0+        "select 1"+        []+        True+        (HasqlErrors.UnexpectedRowCountStatementError lo hi actual)+    )++-- | Connection-level samples (acquisition and connect failures).+connectionSamples :: [(String, PgmqRuntimeError)]+connectionSamples =+  [ ("networking connection error", PgmqConnectionError (HasqlErrors.NetworkingConnectionError "refused")),+    ("authentication error", PgmqConnectionError (HasqlErrors.AuthenticationConnectionError "bad password")),+    ("compatibility error", PgmqConnectionError (HasqlErrors.CompatibilityConnectionError "version mismatch")),+    ("other connection error", PgmqConnectionError (HasqlErrors.OtherConnectionError "libpq says no"))+  ]++-- | Every sample error this module constructs, for whole-table properties.+allSamples :: [(String, PgmqRuntimeError)]+allSamples =+  [("acquisition timeout", PgmqAcquisitionTimeout)]+    <> connectionSamples+    <> [ ("connection-drop session error", PgmqSessionError (HasqlErrors.ConnectionSessionError "dropped")),+         ("driver session error", PgmqSessionError (HasqlErrors.DriverSessionError "bug")),+         ("missing-types session error", PgmqSessionError (HasqlErrors.MissingTypesSessionError HashSet.empty)),+         ("empty-SQLSTATE statement error", lostReplyStatementError),+         ("empty-SQLSTATE script error", scriptError ""),+         ("row count 1 1 0", rowCountError 1 1 0),+         ("row count 1 1 1", rowCountError 1 1 1),+         ("row count 1 1 2", rowCountError 1 1 2)+       ]+    <> [("statement " <> show code, serverStatementError code) | (code, _) <- transientStates <> permanentStates]+    <> [("script " <> show code, scriptError code) | (code, _) <- transientStates <> permanentStates]+ -- | SQLSTATEs that retries exist for: they arrive as server errors inside -- 'StatementSessionError' and must classify transient. transientStates :: [(Text, String)]@@ -115,20 +179,58 @@                | (code, name) <- permanentStates                ]         ),-      testCase "row-count decode failure is not transient" $-        assertBool-          "expected permanent"-          ( not $-              isTransient-                ( PgmqSessionError-                    ( HasqlErrors.StatementSessionError-                        1-                        0-                        "select 1"-                        []-                        True-                        (HasqlErrors.UnexpectedRowCountStatementError 1 1 0)-                    )-                )-          )+      testCase "row-count decode failure (1 1 0) is not transient" $+        assertBool "expected permanent" (not (isTransient (rowCountError 1 1 0))),+      testCase "row-count decode failure (1 1 2) is not transient" $+        assertBool "expected permanent" (not (isTransient (rowCountError 1 1 2))),+      testGroup+        "lost replies (BUG-1)"+        [ testCase "statement error with empty SQLSTATE is transient" $+            assertBool "expected transient" (isTransient lostReplyStatementError),+          testCase "script error with empty SQLSTATE is transient" $+            assertBool "expected transient" (isTransient (scriptError "")),+          testCase "stray-result shape (1 1 1) is transient" $+            assertBool "expected transient" (isTransient (rowCountError 1 1 1))+        ],+      testGroup+        "script errors follow the SQLSTATE rule"+        [ testCase "script 57P01 is transient" $+            assertBool "expected transient" (isTransient (scriptError "57P01")),+          testCase "script 23505 is permanent" $+            assertBool "expected permanent" (not (isTransient (scriptError "23505")))+        ],+      testGroup+        "isAmbiguousReply"+        ( [ testCase "empty-SQLSTATE statement error is ambiguous" $+              assertBool "expected ambiguous" (isAmbiguousReply lostReplyStatementError),+            testCase "empty-SQLSTATE script error is ambiguous" $+              assertBool "expected ambiguous" (isAmbiguousReply (scriptError "")),+            testCase "stray-result shape (1 1 1) is ambiguous" $+              assertBool "expected ambiguous" (isAmbiguousReply (rowCountError 1 1 1))+          ]+            <> [ testCase (name <> " is not ambiguous") $+                   assertBool "expected unambiguous" (not (isAmbiguousReply err))+               | (name, err) <-+                   [ ("statement \"57P01\"", serverStatementError "57P01"),+                     ("statement \"23505\"", serverStatementError "23505"),+                     ("script \"57P01\"", scriptError "57P01"),+                     ("row count 1 1 0", rowCountError 1 1 0),+                     ("row count 1 1 2", rowCountError 1 1 2),+                     ("connection-drop session error", PgmqSessionError (HasqlErrors.ConnectionSessionError "dropped")),+                     ("acquisition timeout", PgmqAcquisitionTimeout),+                     ("driver session error", PgmqSessionError (HasqlErrors.DriverSessionError "bug")),+                     ("missing-types session error", PgmqSessionError (HasqlErrors.MissingTypesSessionError HashSet.empty))+                   ]+                     <> connectionSamples+               ]+            <> [ testCase "every ambiguous reply is transient" $+                   mapM_+                     ( \(name, err) ->+                         assertBool+                           (name <> " is ambiguous but not transient")+                           (not (isAmbiguousReply err) || isTransient err)+                     )+                     allSamples+               ]+        )     ]

@@ -0,0 +1,324 @@+{-# LANGUAGE DuplicateRecordFields #-}+{-# LANGUAGE OverloadedStrings #-}++-- | BUG-1: a connection lost while a statement is in flight must classify as+-- transient, whichever hasql constructor it arrives in.+--+-- Drives three real faults against a dedicated PostgreSQL (never the+-- suite-shared one) while a 'readWithPoll' is blocked server-side, records+-- what surfaced, and asserts that 'isTransient' says retry and that the same+-- pool completes a later send.+--+-- All work happens inside one 'withResource' acquisition; the test cases are+-- pure assertions over the recorded observations, because tasty runs cases+-- concurrently and a sequence of faults must not be split across them.+module DisconnectSpec (tests) where++import Control.Concurrent (forkIO, threadDelay)+import Control.Concurrent.MVar (newEmptyMVar, putMVar, takeMVar)+import Control.Exception (SomeException, bracket, try)+import Data.Aeson qualified as Aeson+import Data.IORef (IORef, newIORef, readIORef, writeIORef)+import Data.Int (Int32, Int64)+import Data.Text qualified as T+import Data.Word (Word32)+import Effectful (Eff, IOE, runEff)+import Effectful.Error.Static (Error, runError)+import EphemeralDb (ephemeralConfig, installPgmqNative)+import EphemeralPg qualified as Pg++-- 'shutdownMode' names a field of both Pg.Config and Pg.Database, so the record+-- update below needs the selector from the module that defines only Database.+import EphemeralPg.Database qualified as PgDb+import Hasql.Decoders qualified as D+import Hasql.Encoders qualified as E+import Hasql.Errors qualified as HasqlErrors+import Hasql.Pool qualified as Pool+import Hasql.Pool.Config qualified as PoolConfig+import Hasql.Session (Session)+import Hasql.Session qualified as Session+import Hasql.Statement (Statement, preparable)+import Pgmq.Effectful+  ( MessageBody (..),+    Pgmq,+    PgmqRuntimeError (..),+    QueueName,+    ReadWithPollMessage (..),+    SendMessage (..),+    createQueue,+    isAmbiguousReply,+    isTransient,+    parseQueueName,+    readWithPoll,+    runPgmq,+    sendMessage,+  )+import System.Posix.Signals (sigKILL, signalProcess)+import System.Random (randomRIO)+import Test.Tasty (TestTree, testGroup, withResource)+import Test.Tasty.HUnit (assertBool, assertFailure, testCase)++-- | What one fault produced.+data FaultObservation = FaultObservation+  { -- | The error the in-flight poll surfaced; Nothing means it completed.+    obsError :: !(Maybe PgmqRuntimeError),+    -- | Attempts the same pool needed to complete a send afterwards.+    obsRecoveryAttempts :: !Int,+    -- | Errors seen on the way to recovery, in order.+    obsRecoveryErrors :: ![PgmqRuntimeError],+    -- | Whether the send eventually succeeded within the attempt budget.+    obsRecovered :: !Bool+  }++data DisconnectObservations = DisconnectObservations+  { obsTerminate :: !FaultObservation,+    obsKill :: !FaultObservation,+    obsRestart :: !FaultObservation+  }++tests :: TestTree+tests =+  withResource runDisconnectCycle (const (pure ())) $ \getObs ->+    testGroup+      "DisconnectSpec (BUG-1)"+      [ faultCases "backend termination" (obsTerminate <$> getObs),+        faultCases "backend SIGKILL" (obsKill <$> getObs),+        faultCases "immediate restart" (obsRestart <$> getObs),+        testCase "SIGKILL reply is ambiguous" $ do+          obs <- obsKill <$> getObs+          case obsError obs of+            Nothing -> pure ()+            Just err -> assertBool ("expected an ambiguous reply, got " <> show err) (isAmbiguousReply err)+      ]++faultCases :: String -> IO FaultObservation -> TestTree+faultCases label getObs =+  testGroup+    label+    [ testCase "surfaces an error" $ do+        obs <- getObs+        case obsError obs of+          Nothing -> assertFailure "the in-flight poll completed; the fault did not land"+          Just _ -> pure (),+      testCase "error is a disconnect shape" $ do+        obs <- getObs+        case obsError obs of+          Nothing -> pure ()+          Just err -> assertBool ("unexpected shape: " <> show err) (isDisconnectShape err),+      testCase "error is transient" $ do+        obs <- getObs+        case obsError obs of+          Nothing -> pure ()+          Just err -> assertBool ("expected transient, got " <> show err) (isTransient err),+      testCase "ambiguous replies are transient" $ do+        obs <- getObs+        case obsError obs of+          Nothing -> pure ()+          Just err ->+            assertBool+              ("ambiguous but not transient: " <> show err)+              (not (isAmbiguousReply err) || isTransient err),+      testCase "same pool recovers" $ do+        obs <- getObs+        assertBool+          ( "pool did not recover within "+              <> show (obsRecoveryAttempts obs)+              <> " attempts; errors seen: "+              <> show (obsRecoveryErrors obs)+          )+          (obsRecovered obs)+    ]++-- | The values a lost connection is known to arrive in (see design note 017).+isDisconnectShape :: PgmqRuntimeError -> Bool+isDisconnectShape = \case+  PgmqSessionError (HasqlErrors.ConnectionSessionError _) -> True+  PgmqSessionError (HasqlErrors.StatementSessionError _ _ _ _ _ statementError) ->+    case statementError of+      HasqlErrors.ServerStatementError (HasqlErrors.ServerError code _ _ _ _) ->+        code `elem` ["", "57P01", "57P02"]+      HasqlErrors.UnexpectedRowCountStatementError 1 1 1 -> True+      _ -> False+  _ -> False++-- The fault cycle -------------------------------------------------------------++-- | Start a dedicated PostgreSQL, run the three faults in order against one+-- size-1 pool, and record what each produced. Every resource is released+-- before this returns.+runDisconnectCycle :: IO DisconnectObservations+runDisconnectCycle = do+  db0 <- startOrFail+  ref <- newIORef db0+  bracket (pure ref) (\r -> readIORef r >>= Pg.stop) $ \dbRef -> do+    installPgmqNative (Pg.connectionSettings db0)+    -- The poll queue stays empty so every poll blocks server-side; recovery+    -- sends go to a separate queue so they never satisfy a later poll.+    pollQueue <- genQueueName "disconnect_poll_"+    sendQueue <- genQueueName "disconnect_send_"+    bracket (acquirePool db0) Pool.release $ \pool -> do+      runOrFail pool (createQueue pollQueue >> createQueue sendQueue)++      terminate <- faultRound "backend termination" dbRef pool pollQueue sendQueue $ \pid -> do+        db <- readIORef dbRef+        bracket (acquirePool db) Pool.release $ \control ->+          () <$ session control (Session.statement pid terminateBackend)++      kill <- faultRound "backend SIGKILL" dbRef pool pollQueue sendQueue $ \pid ->+        signalProcess sigKILL (fromIntegral pid)++      restart <- faultRound "immediate restart" dbRef pool pollQueue sendQueue $ \_pid -> do+        db <- readIORef dbRef+        db' <- crashAndRecover db+        writeIORef dbRef db'++      pure DisconnectObservations {obsTerminate = terminate, obsKill = kill, obsRestart = restart}++-- | Read the pool connection's backend pid, block it in 'readWithPoll', inject+-- the fault, collect what the poll surfaced, and then measure how many sends+-- the same pool needs to succeed again.+faultRound ::+  String ->+  IORef Pg.Database ->+  Pool.Pool ->+  QueueName ->+  QueueName ->+  (Int32 -> IO ()) ->+  IO FaultObservation+faultRound label dbRef pool pollQueue sendQueue inject = do+  -- ephemeral-pg runs PostgreSQL with fsync and synchronous_commit off, so a+  -- crash would otherwise discard the queues still sitting in the WAL buffers.+  session pool (Session.script "checkpoint")+  pid <- session pool (Session.statement () backendPid)+  outcomeVar <- newEmptyMVar+  _ <- forkIO $ do+    result <- try @SomeException (runPoll pool pollQueue)+    putMVar outcomeVar result+  db <- readIORef dbRef+  awaitPolling db pid 200+  inject pid+  outcome <- takeMVar outcomeVar+  err <- case outcome of+    Left exc -> assertFailure (label <> ": poll thread threw " <> show exc)+    Right (Left runtimeErr) -> pure (Just runtimeErr)+    Right (Right ()) -> pure Nothing+  (attempts, errors, recovered) <- recover pool sendQueue 100+  pure+    FaultObservation+      { obsError = err,+        obsRecoveryAttempts = attempts,+        obsRecoveryErrors = errors,+        obsRecovered = recovered+      }++runPoll :: Pool.Pool -> QueueName -> IO (Either PgmqRuntimeError ())+runPoll pool queue =+  runPgmqIO pool . (() <$) . readWithPoll $+    ReadWithPollMessage+      { queueName = queue,+        delay = 30,+        batchSize = Just 1,+        maxPollSeconds = 15,+        pollIntervalMs = 100,+        conditional = Nothing+      }++-- | Wait until the server shows the given backend busy in @read_with_poll@.+-- The attacked pool's only connection is busy, so this opens its own+-- short-lived control pool each time (the SIGKILL round restarts every+-- backend, so nothing is shared across rounds).+awaitPolling :: Pg.Database -> Int32 -> Int -> IO ()+awaitPolling db pid budget =+  bracket (acquirePool db) Pool.release $ \control -> go control budget+  where+    go _ 0 = assertFailure ("backend " <> show pid <> " never showed read_with_poll as active")+    go control n = do+      count <- Pool.use control (Session.statement pid activePollCount)+      case count of+        Right 1 -> pure ()+        _ -> threadDelay 50_000 >> go control (n - 1)++-- | Send one message through the attacked pool, retrying while the error is+-- transient, with a 100 ms pause, up to the attempt budget. Returns the+-- attempts used, every error seen in order, and whether a send succeeded. A+-- permanent error stops the loop.+recover :: Pool.Pool -> QueueName -> Int -> IO (Int, [PgmqRuntimeError], Bool)+recover pool queue budget = go 1 []+  where+    go attempt seen = do+      result <-+        runPgmqIO pool $+          () <$ sendMessage SendMessage {queueName = queue, messageBody = MessageBody (Aeson.String "probe"), delay = Nothing}+      case result of+        Right () -> pure (attempt, reverse seen, True)+        Left err+          | attempt >= budget || not (isTransient err) -> pure (attempt, reverse (err : seen), False)+          | otherwise -> threadDelay 100_000 >> go (attempt + 1) (err : seen)++-- | Stop PostgreSQL with SIGQUIT and start it again on the same data directory+-- and port. Retries once, then fails loudly.+crashAndRecover :: Pg.Database -> IO Pg.Database+crashAndRecover db = do+  let crashing = db {PgDb.shutdownMode = Pg.ShutdownImmediate}+  first <- Pg.restart crashing+  case first of+    Right db' -> pure db'+    Left _ -> do+      second <- Pg.restart crashing+      case second of+        Right db' -> pure db'+        Left err -> assertFailure $ "could not restart PostgreSQL after crash: " <> show err++-- Database plumbing -----------------------------------------------------------++startOrFail :: IO Pg.Database+startOrFail = do+  config <- ephemeralConfig+  result <- Pg.startCached config Pg.defaultCacheConfig+  case result of+    Left err -> assertFailure $ "could not start a dedicated PostgreSQL: " <> show err+    Right db -> pure db++-- | Size 1 is essential: it guarantees the backend pid read before the fault+-- belongs to the connection the poll uses.+acquirePool :: Pg.Database -> IO Pool.Pool+acquirePool db =+  Pool.acquire $+    PoolConfig.settings+      [ PoolConfig.size 1,+        PoolConfig.staticConnectionSettings (Pg.connectionSettings db)+      ]++runPgmqIO :: Pool.Pool -> Eff '[Pgmq, Error PgmqRuntimeError, IOE] a -> IO (Either PgmqRuntimeError a)+runPgmqIO pool action = either (Left . snd) Right <$> runEff (runError @PgmqRuntimeError (runPgmq pool action))++runOrFail :: Pool.Pool -> Eff '[Pgmq, Error PgmqRuntimeError, IOE] a -> IO a+runOrFail pool action =+  runPgmqIO pool action >>= either (\err -> assertFailure ("pgmq operation failed: " <> show err)) pure++session :: Pool.Pool -> Session a -> IO a+session pool s = do+  result <- Pool.use pool s+  case result of+    Left err -> assertFailure $ "Session failed: " <> show err+    Right a -> pure a++genQueueName :: T.Text -> IO QueueName+genQueueName prefix = do+  suffix <- randomRIO (10000 :: Word32, 99999)+  case parseQueueName (prefix <> T.pack (show suffix)) of+    Left err -> error $ "Failed to generate queue name: " <> show err+    Right qn -> pure qn++backendPid :: Statement () Int32+backendPid = preparable "select pg_backend_pid()" E.noParams (D.singleRow (D.column (D.nonNullable D.int4)))++terminateBackend :: Statement Int32 Bool+terminateBackend = preparable "select pg_terminate_backend($1)" (E.param (E.nonNullable E.int4)) (D.singleRow (D.column (D.nonNullable D.bool)))++activePollCount :: Statement Int32 Int64+activePollCount =+  preparable+    "select count(*)::int8 from pg_stat_activity where pid = $1 and state = 'active' and query like '%read_with_poll%'"+    (E.param (E.nonNullable E.int4))+    (D.singleRow (D.column (D.nonNullable D.int8)))

@@ -9,11 +9,15 @@ module EphemeralDb   ( withPgmqDb,     withPgmqPool,+    ephemeralConfig,+    ensureOwnerOnlyRoot,+    installPgmqNative,     StartError,   ) where -import Control.Monad (filterM, when)+import Control.Monad (filterM, unless, when)+import Data.Bits ((.&.)) import Data.List.NonEmpty (NonEmpty (..)) import Data.Monoid (Last (..)) import Data.Text.IO qualified as TextIO@@ -30,12 +34,16 @@     defaultConfig,     withCachedConfig,   )+import Hasql.Connection.Settings qualified as ConnectionSettings import Hasql.Pool qualified as Pool import Hasql.Pool.Config qualified as PoolConfig import Hasql.Session qualified as Session import Pgmq.Migration qualified as Migration-import System.Directory (createDirectoryIfMissing, doesFileExist)+import System.Directory (doesFileExist) import System.Environment (lookupEnv)+import System.IO.Error (catchIOError, isAlreadyExistsError)+import System.Posix.Directory qualified as Posix+import System.Posix.Files (deviceID, fileID, fileMode, fileOwner, getSymbolicLinkStatus, isDirectory, setFileMode) import System.Posix.User (getEffectiveUserID)  -- | Root directory for ephemeral PostgreSQL clusters.@@ -53,11 +61,31 @@   uid <- getEffectiveUserID   pure ("/tmp/ephpg-pgmq-hs-" <> show uid) +-- | Establish a real effective-uid-owned root without deleting its contents.+ensureOwnerOnlyRoot :: FilePath -> IO ()+ensureOwnerOnlyRoot root = do+  catchIOError (Posix.createDirectory root 0o700) $ \err ->+    unless (isAlreadyExistsError err) (ioError err)+  uid <- getEffectiveUserID+  before <- getSymbolicLinkStatus root+  unless (isDirectory before && fileOwner before == uid) $+    ioError (userError ("Unsafe ephemeral PostgreSQL root: " <> root))+  setFileMode root 0o700+  after <- getSymbolicLinkStatus root+  unless+    ( isDirectory after+        && fileOwner after == uid+        && deviceID before == deviceID after+        && fileID before == fileID after+        && fileMode after .&. 0o7777 == 0o700+    )+    (ioError (userError ("Ephemeral PostgreSQL root changed or is not owner-only: " <> root)))+ -- | Cached-startup configuration pinned to 'ephemeralRoot'. ephemeralConfig :: IO Config ephemeralConfig = do   root <- ephemeralRoot-  createDirectoryIfMissing True root+  ensureOwnerOnlyRoot root   pure defaultConfig {temporaryRoot = Last (Just root)}  -- | Run an action with a temporary PostgreSQL database that has the@@ -82,8 +110,8 @@             [] -> error "Missing packaged PGMQ 1.12.0 test fixture"           sql <- TextIO.readFile path           Pool.use pool (Session.script sql) >>= either (error . show) pure-        Nothing -> installNative connSettings-        Just "1.13.0" -> installNative connSettings+        Nothing -> installPgmqNative connSettings+        Just "1.13.0" -> installPgmqNative connSettings         Just invalid -> error ("Invalid PGMQ_TEST_SCHEMA_VERSION: " <> invalid)       required <- (== Just "1") <$> lookupEnv "PGMQ_REQUIRE_PARTMAN"       -- Required runs fail for absent or unusable pg_partman, not just missing metadata.@@ -92,12 +120,15 @@         Left err -> when required (error ("PGMQ_REQUIRE_PARTMAN=1: " <> show err))         Right () -> pure ()       action pool-  where-    installNative connSettings = do-      component <- either (error . ("Invalid PGMQ migration component: " <>) . show) pure Migration.pgmqMigrations-      plan <- either (error . ("Invalid PGMQ migration plan: " <>) . show) pure (migrationPlan (component :| []))-      installResult <- runMigrationPlan defaultRunOptions connSettings plan-      either (error . ("Migration failed: " <>) . show) (const (pure ())) installResult++-- | Apply the native pgmq migration ledger to the database at the given+-- connection settings, failing loudly on any error.+installPgmqNative :: ConnectionSettings.Settings -> IO ()+installPgmqNative connSettings = do+  component <- either (error . ("Invalid PGMQ migration component: " <>) . show) pure Migration.pgmqMigrations+  plan <- either (error . ("Invalid PGMQ migration plan: " <>) . show) pure (migrationPlan (component :| []))+  installResult <- runMigrationPlan defaultRunOptions connSettings plan+  either (error . ("Migration failed: " <>) . show) (const (pure ())) installResult  -- | Alias for 'withPgmqDb' kept for parallelism with pgmq-hasql. withPgmqPool :: (Pool.Pool -> IO a) -> IO (Either StartError a)

@@ -3,8 +3,10 @@ module Main (main) where  import ClassificationSpec qualified-import EphemeralDb (withPgmqPool)+import DisconnectSpec qualified+import EphemeralDb (ensureOwnerOnlyRoot, withPgmqPool) import PlainInterpreterSpec qualified+import RootPermissionsSpec qualified import Test.Tasty (defaultMain, testGroup) import TracedInterpreterSpec qualified import UmbrellaExportsSpec ()@@ -15,9 +17,11 @@     let tree =           testGroup             "pgmq-effectful"-            [ ClassificationSpec.tests,+            [ RootPermissionsSpec.tests ensureOwnerOnlyRoot,+              ClassificationSpec.tests,               PlainInterpreterSpec.tests pool,-              TracedInterpreterSpec.tests pool+              TracedInterpreterSpec.tests pool,+              DisconnectSpec.tests             ]     defaultMain tree   case result of

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

@@ -521,7 +521,57 @@                        Left (_, PgmqSessionError err) -> assertBool "server rejection is retained" (T.isInfixOf (if stock then "42883" else "premake must be at least 1") (T.pack (show err)))                        other -> assertFailure ("expected session error, got " <> show other)                  )-                 [0, -1]+                 [0, -1],+           testCase "non-destructive inspection reads" $+             withSemconvOptIn "messaging/dup,database/dup" $+               isolated $ \pool -> do+                 (tracer, _, spansRef) <- setupTracer+                 let run :: Eff '[Eff.Pgmq, Error PgmqRuntimeError, IOE] a -> IO a+                     run action = assertRight =<< runEff (runError @PgmqRuntimeError ((if traced then runPgmqTraced pool tracer else runPgmq pool) action))+                 queue <- mkUniqueQueue "inspect"+                 let q = queueNameToText queue+                 run (Eff.createQueue queue)+                 ids <- mapM (\i -> run (Eff.sendMessage (SendMessage queue (MessageBody (object ["i" .= (i :: Int)])) Nothing))) [1 .. 3]+                 (archivedId, liveId) <- case ids of+                   [a, b, _] -> pure (a, b)+                   other -> assertFailure ("expected three ids, got " <> show other)+                 archivedOk <- run (Eff.archiveMessage (MessageQuery queue archivedId))+                 assertBool "archive succeeded" archivedOk+                 page <- run (Eff.peekMessages (Types.PeekMessages q Nothing 10))+                 assertEqual "two live messages" (drop 1 ids) (map Pgmq.messageId (V.toList page))+                 archived <- run (Eff.peekArchivedMessages (Types.PeekMessages q Nothing 10))+                 assertEqual "one archived message" [archivedId] (map (Pgmq.messageId . Pgmq.archivedMessage) (V.toList archived))+                 found <- run (Eff.lookupMessage (Types.LookupMessage q liveId))+                 assertEqual "lookup finds a live message" (Just liveId) (fmap Pgmq.messageId found)+                 missing <- run (Eff.lookupArchivedMessage (Types.LookupMessage q liveId))+                 assertEqual "a live message is not in the archive" Nothing (fmap (Pgmq.messageId . Pgmq.archivedMessage) missing)+                 metrics <- run (Eff.queueMetricsUnvalidated q)+                 assertEqual "lenient metrics" 2 (Types.queueLength metrics)+                 -- The peeks leased nothing: a real read now returns both, and+                 -- read_ct is one because only this read incremented it.+                 leased <- run (Eff.readMessage (ReadMessage queue 30 (Just 10) Nothing))+                 assertEqual "both messages still available" (drop 1 ids) (map Pgmq.messageId (V.toList leased))+                 assertEqual "read_ct counts only the real read" [1, 1] (map Pgmq.readCount (V.toList leased))+                 when traced $ do+                   spans <- readIORef spansRef+                   peeks <- spansWithFirstWord "pgmq.peek" spans+                   peekSpan <- singleSpan peeks "peek"+                   spanName peekSpan >>= assertEqual "peek span name" ("pgmq.peek " <> q)+                   case OTel.spanKind peekSpan of+                     OTel.Internal -> pure ()+                     other -> assertFailure ("expected Internal: " <> show other)+                   assertAttrText peekSpan "db.operation" "pgmq.peek"+                   assertAttrText peekSpan "db.operation.name" "pgmq.peek"+                   assertAttrText peekSpan "messaging.destination.name" q+                   assertNoAttr peekSpan "messaging.operation"+                   archivePeeks <- spansWithFirstWord "pgmq.peek_archive" spans+                   _ <- singleSpan archivePeeks "archive peek"+                   lookups <- spansWithFirstWord "pgmq.lookup_message" spans+                   lookupSpan <- singleSpan lookups "lookup"+                   assertAttrText lookupSpan "messaging.message.id" (T.pack (show (Pgmq.unMessageId liveId)))+                   archivedLookups <- spansWithFirstWord "pgmq.lookup_archived_message" spans+                   _ <- singleSpan archivedLookups "archived lookup"+                   pure ()          ]  session :: Pool.Pool -> Session.Session a -> IO a

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