diff --git a/CHANGELOG.md b/CHANGELOG.md --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,44 @@ # Revision history for pgmq-effectful +## 0.5.0.0 -- 2026-08-06 + +### Breaking Changes + +* `changeVisibilityTimeout` and `setVisibilityTimeoutAt` now return `Maybe Message` + instead of `Message`. `pgmq.set_vt` is `RETURNS SETOF` and yields zero rows when the + target message no longer exists (already deleted, archived, or popped), which the + single-row decoder turned into an `UnexpectedRowCountStatementError` indistinguishable + from infrastructure failure. Callers that used the result must now handle `Nothing`; + callers that discarded it compile unchanged. The batch variants are unaffected. +* The `Pgmq` effect GADT gains the `ListQueuesUnvalidated` and `ListFifoIndexQueueNames` + constructors. Custom interpreters that match exhaustively must handle them; both stock + interpreters already do. + +### New Features + +* `listQueuesUnvalidated` reads the queue listing with names decoded as `Text` + (`UnvalidatedQueue`), so a queue created by another client under a name + `parseQueueName` rejects does not fail the whole listing. Its traced span keeps the + `pgmq.list_queues` name, because the SQL function invoked is the same. +* `listFifoIndexQueueNames` reports which queues already carry a `q__fifo_idx`, read from + the `pg_indexes` catalog view. Its traced span is named `pgmq.list_fifo_indexes` after + this library's own operation, since no `pgmq.*` SQL function backs it. + +### Bug Fixes + +* `isTransient` now classifies retry-worthy server errors as transient. Serialization + failures (40001), deadlocks (40P01), lock timeouts (55P03), server shutdown and recovery + (57P01/57P02/57P03), and resource exhaustion (class 53) all arrive as server errors + inside `StatementSessionError`, which previously mapped to permanent unconditionally — + so retry loops gated on `isTransient` failed fast on exactly the errors retries exist + for. Every other statement error, including decode and row-count mismatches, remains + permanent. The whitelist is pinned in both directions by tests and recorded in + `docs/design/017-transient-error-classification.md`. + +### Other Changes + +* Bumped `pgmq-core` and `pgmq-hasql` dependency bounds to `>=0.5 && <0.6`. + ## 0.4.0.1 -- 2026-07-14 * Version bump only — coordinated release with pgmq-migration 0.4.0.1. 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.4.0.1 +version: 0.5.0.0 synopsis: Effectful effects for PGMQ (PostgreSQL Message Queue) description: Effectful effects and interpreters for pgmq-hs, a Haskell client @@ -57,8 +57,8 @@ , hs-opentelemetry-api >=1.0 && <2 , hs-opentelemetry-semantic-conventions >=1.40 && <2 , http-types >=0.12 && <0.13 - , pgmq-core >=0.4 && <0.5 - , pgmq-hasql >=0.4 && <0.5 + , pgmq-core >=0.5 && <0.6 + , pgmq-hasql >=0.5 && <0.6 , text >=2.0 && <2.2 , unliftio >=0.2 && <0.3 , unordered-containers >=0.2 && <0.3 @@ -101,9 +101,9 @@ , hs-opentelemetry-sdk >=1.0 && <2 , hs-opentelemetry-semantic-conventions >=1.40 && <2 , pg-migrate - , pgmq-core >=0.4 && <0.5 + , pgmq-core >=0.5 && <0.6 , pgmq-effectful - , pgmq-migration >=0.4 && <0.5 + , pgmq-migration >=0.5 && <0.6 , random ^>=1.2 , tasty ^>=1.5 , tasty-hunit ^>=0.10 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 @@ -76,12 +76,15 @@ -- * Queue Observability listQueues, + listQueuesUnvalidated, + listFifoIndexQueueNames, queueMetrics, allQueueMetrics, ) where import Data.Int (Int32, Int64) +import Data.Text (Text) import Data.Vector (Vector) import Effectful (Dispatch (..), DispatchOf, Eff, Effect, (:>)) import Effectful.Dispatch.Dynamic (send) @@ -129,6 +132,7 @@ TopicBinding, TopicPattern, TopicSendResult, + UnvalidatedQueue, ) -- | Effect for pgmq message queue operations. @@ -158,10 +162,10 @@ ArchiveMessage :: MessageQuery -> Pgmq m Bool BatchArchiveMessages :: BatchMessageQuery -> Pgmq m [MessageId] DeleteAllMessagesFromQueue :: QueueName -> Pgmq m Int64 - ChangeVisibilityTimeout :: VisibilityTimeoutQuery -> Pgmq m Message + ChangeVisibilityTimeout :: VisibilityTimeoutQuery -> Pgmq m (Maybe Message) BatchChangeVisibilityTimeout :: BatchVisibilityTimeoutQuery -> Pgmq m (Vector Message) -- Timestamp-based VT (pgmq 1.10.0+) - SetVisibilityTimeoutAt :: VisibilityTimeoutAtQuery -> Pgmq m Message + SetVisibilityTimeoutAt :: VisibilityTimeoutAtQuery -> Pgmq m (Maybe Message) BatchSetVisibilityTimeoutAt :: BatchVisibilityTimeoutAtQuery -> Pgmq m (Vector Message) ReadWithPoll :: ReadWithPollMessage -> Pgmq m (Vector Message) Pop :: PopMessage -> Pgmq m (Vector Message) @@ -191,6 +195,8 @@ UpdateNotifyInsert :: UpdateNotifyInsert -> Pgmq m () -- Queue Observability ListQueues :: Pgmq m [Queue] + ListQueuesUnvalidated :: Pgmq m [UnvalidatedQueue] + ListFifoIndexQueueNames :: Pgmq m [Text] QueueMetrics :: QueueName -> Pgmq m QueueMetrics AllQueueMetrics :: Pgmq m [QueueMetrics] @@ -272,14 +278,17 @@ deleteAllMessagesFromQueue :: (Pgmq :> es) => QueueName -> Eff es Int64 deleteAllMessagesFromQueue = send . DeleteAllMessagesFromQueue -changeVisibilityTimeout :: (Pgmq :> es) => VisibilityTimeoutQuery -> Eff es Message +-- | Returns Nothing when the message no longer exists (already deleted, archived, +-- or popped) rather than throwing. +changeVisibilityTimeout :: (Pgmq :> es) => VisibilityTimeoutQuery -> Eff es (Maybe Message) changeVisibilityTimeout = send . ChangeVisibilityTimeout batchChangeVisibilityTimeout :: (Pgmq :> es) => BatchVisibilityTimeoutQuery -> Eff es (Vector Message) batchChangeVisibilityTimeout = send . BatchChangeVisibilityTimeout --- | Set visibility timeout to an absolute timestamp (pgmq 1.10.0+) -setVisibilityTimeoutAt :: (Pgmq :> es) => VisibilityTimeoutAtQuery -> Eff es Message +-- | Set visibility timeout to an absolute timestamp (pgmq 1.10.0+). +-- Returns Nothing when the message no longer exists. +setVisibilityTimeoutAt :: (Pgmq :> es) => VisibilityTimeoutAtQuery -> Eff es (Maybe Message) setVisibilityTimeoutAt = send . SetVisibilityTimeoutAt -- | Batch set visibility timeout to an absolute timestamp (pgmq 1.10.0+) @@ -380,6 +389,17 @@ listQueues :: (Pgmq :> es) => Eff es [Queue] listQueues = send ListQueues + +-- | Like 'listQueues' but with queue names left unvalidated, so a queue +-- created by another client under a name 'Pgmq.Types.parseQueueName' rejects +-- does not fail the whole listing. +listQueuesUnvalidated :: (Pgmq :> es) => Eff es [UnvalidatedQueue] +listQueuesUnvalidated = send ListQueuesUnvalidated + +-- | Queue names that already have their FIFO headers index. Reads the +-- @pg_indexes@ catalog view; pgmq exposes no index-existence function. +listFifoIndexQueueNames :: (Pgmq :> es) => Eff es [Text] +listFifoIndexQueueNames = send ListFifoIndexQueueNames queueMetrics :: (Pgmq :> es) => QueueName -> Eff es QueueMetrics queueMetrics = send . QueueMetrics 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 @@ -13,6 +13,8 @@ where import Control.Exception (Exception) +import Data.Text (Text) +import Data.Text qualified as T import Effectful (Eff, IOE, (:>)) import Effectful qualified import Effectful.Dispatch.Dynamic (interpret) @@ -53,10 +55,14 @@ -- | Is this error plausibly transient — i.e., worth retrying? -- -- Returns 'True' for acquisition timeouts, networking connection errors, --- uncategorized libpq connection errors, and session-level connection --- drops. All other errors (authentication failure, compatibility --- mismatches, missing types, statement errors, driver bugs) are treated --- as permanent. +-- 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. -- -- Note: 'HasqlErrors.OtherConnectionError' is classed as transient here -- despite hasql's documentation calling it \"not transient by default\", @@ -72,11 +78,28 @@ HasqlErrors.OtherConnectionError _ -> True PgmqSessionError e -> case e of HasqlErrors.ConnectionSessionError _ -> True - HasqlErrors.StatementSessionError {} -> False + HasqlErrors.StatementSessionError _ _ _ _ _ statementError -> + case statementError of + HasqlErrors.ServerStatementError (HasqlErrors.ServerError code _ _ _ _) -> + isTransientSqlState code + _ -> False HasqlErrors.ScriptSessionError {} -> False HasqlErrors.MissingTypesSessionError _ -> False HasqlErrors.DriverSessionError _ -> False +-- | 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 +-- exist for. Everything else reported by the server is permanent for retry +-- purposes. +isTransientSqlState :: Text -> Bool +isTransientSqlState code = + code `elem` ["40001", "40P01", "55P03", "57P01", "57P02", "57P03"] + || "53" `T.isPrefixOf` code + -- | Legacy error type. Retained for one release cycle; migrate to -- 'PgmqRuntimeError'. Will be removed in pgmq-effectful 0.3.0. newtype PgmqError = PgmqPoolError UsageError @@ -149,6 +172,8 @@ UpdateNotifyInsert params -> runSession pool $ Sessions.updateNotifyInsert params -- Queue Observability ListQueues -> runSession pool Sessions.listQueues + ListQueuesUnvalidated -> runSession pool Sessions.listQueuesUnvalidated + ListFifoIndexQueueNames -> runSession pool Sessions.listFifoIndexQueueNames QueueMetrics q -> runSession pool $ Sessions.queueMetrics q AllQueueMetrics -> runSession pool Sessions.allQueueMetrics 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 @@ -333,6 +333,14 @@ ListQueues -> withTracedOp config pool (defaultOpInfo "pgmq.list_queues" OTel.Internal) $ Sessions.listQueues + ListQueuesUnvalidated -> + withTracedOp config pool (defaultOpInfo "pgmq.list_queues" OTel.Internal) $ + Sessions.listQueuesUnvalidated + -- No pgmq function backs this one: it reads the pg_indexes catalog view, so + -- the span carries this library's own label rather than a pgmq.* name. + ListFifoIndexQueueNames -> + withTracedOp config pool (defaultOpInfo "pgmq.list_fifo_indexes" OTel.Internal) $ + Sessions.listFifoIndexQueueNames QueueMetrics q -> withTracedOp config pool (queueOp "pgmq.metrics" OTel.Internal q) $ Sessions.queueMetrics q diff --git a/test/ClassificationSpec.hs b/test/ClassificationSpec.hs --- a/test/ClassificationSpec.hs +++ b/test/ClassificationSpec.hs @@ -3,6 +3,7 @@ module ClassificationSpec (tests) where import Data.HashSet qualified as HashSet +import Data.Text (Text) import Hasql.Errors qualified as HasqlErrors import Pgmq.Effectful ( PgmqRuntimeError (..), @@ -11,6 +12,46 @@ import Test.Tasty (TestTree, testGroup) import Test.Tasty.HUnit (assertBool, testCase) +-- | A server-reported statement error with the given SQLSTATE, wrapped the way +-- it actually arrives at 'isTransient': inside 'StatementSessionError' +-- (statement count, index, SQL, params, prepared flag) around +-- 'ServerStatementError' around 'ServerError', whose first field is the +-- five-character SQLSTATE. +serverStatementError :: Text -> PgmqRuntimeError +serverStatementError code = + PgmqSessionError + ( HasqlErrors.StatementSessionError + 1 + 0 + "select 1" + [] + True + (HasqlErrors.ServerStatementError (HasqlErrors.ServerError code "boom" Nothing Nothing Nothing)) + ) + +-- | SQLSTATEs that retries exist for: they arrive as server errors inside +-- 'StatementSessionError' and must classify transient. +transientStates :: [(Text, String)] +transientStates = + [ ("40001", "serialization_failure"), + ("40P01", "deadlock_detected"), + ("55P03", "lock_not_available"), + ("57P01", "admin_shutdown"), + ("57P02", "crash_shutdown"), + ("57P03", "cannot_connect_now"), + ("53100", "disk_full"), + ("53200", "out_of_memory"), + ("53300", "too_many_connections") + ] + +-- | Genuine statement bugs must stay permanent: retrying cannot fix them. +permanentStates :: [(Text, String)] +permanentStates = + [ ("23505", "unique_violation"), + ("42P01", "undefined_table"), + ("22P02", "invalid_text_representation") + ] + tests :: TestTree tests = testGroup @@ -62,5 +103,32 @@ ( not $ isTransient (PgmqSessionError (HasqlErrors.MissingTypesSessionError HashSet.empty)) + ), + testGroup + "server-reported SQLSTATEs (PGH-10)" + ( [ testCase (show code <> " " <> name <> " is transient") $ + assertBool "expected transient" (isTransient (serverStatementError code)) + | (code, name) <- transientStates + ] + <> [ testCase (show code <> " " <> name <> " is permanent") $ + assertBool "expected permanent" (not (isTransient (serverStatementError code))) + | (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) + ) + ) ) ]