diff --git a/CHANGELOG.md b/CHANGELOG.md --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -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. diff --git a/pgmq-effectful.cabal b/pgmq-effectful.cabal --- a/pgmq-effectful.cabal +++ b/pgmq-effectful.cabal @@ -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, diff --git a/src/Pgmq/Effectful.hs b/src/Pgmq/Effectful.hs --- a/src/Pgmq/Effectful.hs +++ b/src/Pgmq/Effectful.hs @@ -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 (..), diff --git a/src/Pgmq/Effectful/Effect.hs b/src/Pgmq/Effectful/Effect.hs --- a/src/Pgmq/Effectful/Effect.hs +++ b/src/Pgmq/Effectful/Effect.hs @@ -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 diff --git a/src/Pgmq/Effectful/Interpreter.hs b/src/Pgmq/Effectful/Interpreter.hs --- a/src/Pgmq/Effectful/Interpreter.hs +++ b/src/Pgmq/Effectful/Interpreter.hs @@ -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 :: diff --git a/src/Pgmq/Effectful/Interpreter/Traced.hs b/src/Pgmq/Effectful/Interpreter/Traced.hs --- a/src/Pgmq/Effectful/Interpreter/Traced.hs +++ b/src/Pgmq/Effectful/Interpreter/Traced.hs @@ -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 diff --git a/test/ClassificationSpec.hs b/test/ClassificationSpec.hs --- a/test/ClassificationSpec.hs +++ b/test/ClassificationSpec.hs @@ -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 + ] + ) ] diff --git a/test/DisconnectSpec.hs b/test/DisconnectSpec.hs new file mode 100644 --- /dev/null +++ b/test/DisconnectSpec.hs @@ -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))) diff --git a/test/EphemeralDb.hs b/test/EphemeralDb.hs --- a/test/EphemeralDb.hs +++ b/test/EphemeralDb.hs @@ -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) diff --git a/test/Main.hs b/test/Main.hs --- a/test/Main.hs +++ b/test/Main.hs @@ -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 diff --git a/test/RootPermissionsSpec.hs b/test/RootPermissionsSpec.hs new file mode 100644 --- /dev/null +++ b/test/RootPermissionsSpec.hs @@ -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 diff --git a/test/TracedInterpreterSpec.hs b/test/TracedInterpreterSpec.hs --- a/test/TracedInterpreterSpec.hs +++ b/test/TracedInterpreterSpec.hs @@ -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 diff --git a/test/UmbrellaExportsSpec.hs b/test/UmbrellaExportsSpec.hs --- a/test/UmbrellaExportsSpec.hs +++ b/test/UmbrellaExportsSpec.hs @@ -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))