pgmq-effectful 0.4.0.1 → 0.5.0.0
raw patch · 6 files changed
+175/−15 lines, 6 filesdep ~pgmq-coredep ~pgmq-hasqldep ~pgmq-migration
Dependency ranges changed: pgmq-core, pgmq-hasql, pgmq-migration
Files
- CHANGELOG.md +39/−0
- pgmq-effectful.cabal +5/−5
- src/Pgmq/Effectful/Effect.hs +25/−5
- src/Pgmq/Effectful/Interpreter.hs +30/−5
- src/Pgmq/Effectful/Interpreter/Traced.hs +8/−0
- test/ClassificationSpec.hs +68/−0
CHANGELOG.md view
@@ -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_<name>_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.
pgmq-effectful.cabal view
@@ -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
src/Pgmq/Effectful/Effect.hs view
@@ -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
src/Pgmq/Effectful/Interpreter.hs view
@@ -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
src/Pgmq/Effectful/Interpreter/Traced.hs view
@@ -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
test/ClassificationSpec.hs view
@@ -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)+ )+ ) ) ]