shibuya-pgmq-adapter 0.7.0.0 → 0.8.0.0

raw patch · 8 files changed

+226/−11 lines, 8 filesPVP ok

version bump matches the API change (PVP)

API changes (from Hackage documentation)

Files

• CHANGELOG.md +17/−0
• shibuya-pgmq-adapter.cabal +1/−1
• src/Shibuya/Adapter/Pgmq.hs +7/−1
• src/Shibuya/Adapter/Pgmq/Config.hs +25/−0
• src/Shibuya/Adapter/Pgmq/Internal.hs +29/−7
• test/Shibuya/Adapter/Pgmq/ConfigSpec.hs +21/−0
• test/Shibuya/Adapter/Pgmq/IntegrationSpec.hs +3/−0
• test/Shibuya/Adapter/Pgmq/InternalSpec.hs +123/−2

@@ -1,5 +1,22 @@ # Changelog +## 0.8.0.0 — 2026-06-15++### Breaking Changes++- `PgmqAdapterConfig` gained a `pollRetry :: PollRetryConfig` field. Callers+  that construct the config by record literal must add it (or start from a+  smart constructor / default that includes it).+- `pgmqAdapter` now requires `Error PgmqRuntimeError :> es` in its effect row+  so transient poll errors can be caught and retried before being rethrown.++### Bug Fixes++- Transient PGMQ poll errors are retried with bounded exponential backoff+  before the adapter gives up. The default policy makes five total attempts,+  starting at 100ms and capping at five seconds. Permanent errors and+  exhausted retry budgets still surface to shibuya supervision.+ ## 0.7.0.0 — 2026-06-05  Paired with `shibuya-core 0.7.0.0`.

@@ -1,6 +1,6 @@ cabal-version: 3.12 name: shibuya-pgmq-adapter-version: 0.7.0.0+version: 0.8.0.0 synopsis: PGMQ adapter for the Shibuya queue processing framework description:   A Shibuya adapter that integrates with pgmq (PostgreSQL Message Queue)

@@ -75,6 +75,7 @@     -- * Configuration     PgmqAdapterConfig (..),     PollingConfig (..),+    PollRetryConfig (..),     DeadLetterConfig (..),     DeadLetterTarget (..),     FifoConfig (..),@@ -88,6 +89,7 @@     -- * Defaults     defaultConfig,     defaultPollingConfig,+    defaultPollRetryConfig,     defaultPrefetchConfig,      -- * Topic Management (pgmq 1.11.0+)@@ -118,6 +120,8 @@ import Control.Monad.IO.Class (liftIO) import Data.Aeson (Value) import Effectful (Eff, IOE, (:>))+import Effectful.Error.Static (Error)+import Pgmq.Effectful (PgmqRuntimeError) import Pgmq.Effectful.Effect (Pgmq) import Pgmq.Effectful.Effect qualified as PgmqEff import Pgmq.Hasql.Statements.Types qualified as PgmqTypes@@ -141,9 +145,11 @@     FifoConfig (..),     FifoReadStrategy (..),     PgmqAdapterConfig (..),+    PollRetryConfig (..),     PollingConfig (..),     PrefetchConfig (..),     defaultConfig,+    defaultPollRetryConfig,     defaultPollingConfig,     defaultPrefetchConfig,     directDeadLetter,@@ -190,7 +196,7 @@ -- adapter <- pgmqAdapter config -- @ pgmqAdapter ::-  (Pgmq :> es, IOE :> es, Tracing :> es) =>+  (Pgmq :> es, Error PgmqRuntimeError :> es, IOE :> es, Tracing :> es) =>   PgmqAdapterConfig ->   Eff es (Adapter es Value) pgmqAdapter config = do

@@ -5,6 +5,7 @@      -- * Polling Configuration     PollingConfig (..),+    PollRetryConfig (..),      -- * Dead-Letter Queue Configuration     DeadLetterConfig (..),@@ -24,6 +25,7 @@     -- * Defaults     defaultConfig,     defaultPollingConfig,+    defaultPollRetryConfig,     defaultPrefetchConfig,   ) where@@ -45,6 +47,8 @@     batchSize :: !Int32,     -- | Polling configuration     polling :: !PollingConfig,+    -- | Retry policy for transient errors during queue polling+    pollRetry :: !PollRetryConfig,     -- | Optional dead-letter queue configuration     deadLetterConfig :: !(Maybe DeadLetterConfig),     -- | Maximum retries before dead-lettering (default: 3)@@ -58,6 +62,17 @@   }   deriving stock (Show, Eq, Generic) +-- | Retry policy for transient database errors during queue polling.+data PollRetryConfig = PollRetryConfig+  { -- | Total attempts per poll, including the first attempt+    maxAttempts :: !Int,+    -- | Delay before the first retry+    initialBackoff :: !NominalDiffTime,+    -- | Maximum delay between retry attempts+    maxBackoff :: !NominalDiffTime+  }+  deriving stock (Show, Eq, Generic)+ -- | Polling strategy for reading messages. data PollingConfig   = -- | Standard polling with sleep between reads when queue is empty@@ -149,6 +164,15 @@ defaultPollingConfig :: PollingConfig defaultPollingConfig = StandardPolling {pollInterval = 1} +-- | Default retry policy for transient poll errors.+defaultPollRetryConfig :: PollRetryConfig+defaultPollRetryConfig =+  PollRetryConfig+    { maxAttempts = 5,+      initialBackoff = 0.1,+      maxBackoff = 5+    }+ -- | Default prefetch configuration. -- Buffers 4 batches ahead, balancing latency with visibility timeout safety. defaultPrefetchConfig :: PrefetchConfig@@ -173,6 +197,7 @@       visibilityTimeout = 30,       batchSize = 1,       polling = defaultPollingConfig,+      pollRetry = defaultPollRetryConfig,       deadLetterConfig = Nothing,       maxRetries = 3,       fifoConfig = Nothing,

@@ -44,6 +44,8 @@ import Data.Vector (Vector) import Data.Vector qualified as Vector import Effectful (Eff, IOE, (:>))+import Effectful.Error.Static (Error, catchError, throwError)+import Pgmq.Effectful (PgmqRuntimeError, isTransient) import Pgmq.Effectful.Effect   ( Pgmq,     archiveMessage,@@ -79,6 +81,7 @@     FifoConfig (..),     FifoReadStrategy (..),     PgmqAdapterConfig (..),+    PollRetryConfig (..),     PollingConfig (..),   ) import Shibuya.Adapter.Pgmq.Convert@@ -368,11 +371,30 @@ -- Each element is a Vector of messages from a single poll. -- This is the lowest-level stream that handles polling logic. pgmqChunks ::-  (Pgmq :> es, IOE :> es) =>+  (Pgmq :> es, Error PgmqRuntimeError :> es, IOE :> es) =>   PgmqAdapterConfig ->   Stream (Eff es) (Vector Pgmq.Message)-pgmqChunks config = Stream.repeatM poll+pgmqChunks config = Stream.repeatM (pollRetrying 1 initialBackoff)   where+    PollRetryConfig+      { maxAttempts = retryMaxAttempts,+        initialBackoff = initialBackoff,+        maxBackoff = retryMaxBackoff+      } = config.pollRetry++    pollRetrying ::+      (Pgmq :> es, Error PgmqRuntimeError :> es, IOE :> es) =>+      Int ->+      NominalDiffTime ->+      Eff es (Vector Pgmq.Message)+    pollRetrying attempt backoff =+      poll `catchError` \_callStack err ->+        if isTransient err && attempt < retryMaxAttempts+          then do+            liftIO $ threadDelay (nominalToMicros backoff)+            pollRetrying (attempt + 1) (min (backoff * 2) retryMaxBackoff)+          else throwError err+     poll :: (Pgmq :> es, IOE :> es) => Eff es (Vector Pgmq.Message)     poll = case config.fifoConfig of       Nothing -> pollNonFifo@@ -413,7 +435,7 @@ -- Uses Streamly's unfoldEach to expand each Vector into individual elements, -- ensuring ALL messages from each batch are processed (not just the first). pgmqMessages ::-  (Pgmq :> es, IOE :> es) =>+  (Pgmq :> es, Error PgmqRuntimeError :> es, IOE :> es) =>   PgmqAdapterConfig ->   Stream (Eff es) Pgmq.Message pgmqMessages config =@@ -428,7 +450,7 @@ -- This stream polls pgmq and yields Ingested messages. -- Uses unfoldEach to process ALL messages from each batch, not just the first. pgmqSource ::-  (Pgmq :> es, IOE :> es, Tracing :> es) =>+  (Pgmq :> es, Error PgmqRuntimeError :> es, IOE :> es, Tracing :> es) =>   PgmqAdapterConfig ->   Stream (Eff es) (Ingested es Value) pgmqSource config =@@ -443,7 +465,7 @@ -- bufferSize * batchSize * avgProcessingTime < visibilityTimeout to avoid -- messages re-appearing before they're processed. pgmqChunksPrefetch ::-  (Pgmq :> es, IOE :> es) =>+  (Pgmq :> es, Error PgmqRuntimeError :> es, IOE :> es) =>   (StreamP.Config -> StreamP.Config) ->   PgmqAdapterConfig ->   Stream (Eff es) (Vector Pgmq.Message)@@ -454,7 +476,7 @@ -- | Flatten prefetched message chunks into individual messages. -- Like pgmqMessages but with concurrent prefetching of batches. pgmqMessagesPrefetch ::-  (Pgmq :> es, IOE :> es) =>+  (Pgmq :> es, Error PgmqRuntimeError :> es, IOE :> es) =>   (StreamP.Config -> StreamP.Config) ->   PgmqAdapterConfig ->   Stream (Eff es) Pgmq.Message@@ -482,7 +504,7 @@ -- source = pgmqSourceWithPrefetch (StreamP.maxBuffer 2) config -- @ pgmqSourceWithPrefetch ::-  (Pgmq :> es, IOE :> es, Tracing :> es) =>+  (Pgmq :> es, Error PgmqRuntimeError :> es, IOE :> es, Tracing :> es) =>   (StreamP.Config -> StreamP.Config) ->   PgmqAdapterConfig ->   Stream (Eff es) (Ingested es Value)

@@ -8,6 +8,7 @@ spec = do   defaultConfigSpec   defaultPollingConfigSpec+  defaultPollRetryConfigSpec   defaultPrefetchConfigSpec   deadLetterTargetSpec   smartConstructorSpec@@ -34,6 +35,9 @@       StandardPolling interval -> interval `shouldBe` 1       _ -> expectationFailure "Expected StandardPolling" +  it "uses the default poll retry policy" $ do+    config.pollRetry `shouldBe` defaultPollRetryConfig+   it "sets deadLetterConfig to Nothing" $ do     config.deadLetterConfig `shouldBe` Nothing @@ -58,6 +62,23 @@     case defaultPollingConfig of       StandardPolling interval -> interval `shouldBe` 1       _ -> expectationFailure "Expected StandardPolling"++defaultPollRetryConfigSpec :: Spec+defaultPollRetryConfigSpec = describe "defaultPollRetryConfig" $ do+  let PollRetryConfig+        { maxAttempts = retryMaxAttempts,+          initialBackoff = retryInitialBackoff,+          maxBackoff = retryMaxBackoff+        } = defaultPollRetryConfig++  it "retries five total attempts by default" $ do+    retryMaxAttempts `shouldBe` 5++  it "starts with a 100ms backoff" $ do+    retryInitialBackoff `shouldBe` 0.1++  it "caps backoff at five seconds" $ do+    retryMaxBackoff `shouldBe` 5  -- | Tests for defaultPrefetchConfig defaultPrefetchConfigSpec :: Spec

@@ -26,6 +26,7 @@ import Shibuya.Adapter.Pgmq.Config   ( PgmqAdapterConfig (..),     PollingConfig (..),+    defaultPollRetryConfig,   ) import Shibuya.Adapter.Pgmq.Convert (pgmqMessageToEnvelope) import Shibuya.Core.Types (Envelope (..))@@ -310,6 +311,7 @@       visibilityTimeout = 30,       batchSize = 1,       polling = StandardPolling {pollInterval = 0.1}, -- Fast polling for tests+      pollRetry = defaultPollRetryConfig,       deadLetterConfig = Nothing,       maxRetries = 3,       fifoConfig = Nothing,@@ -324,6 +326,7 @@       visibilityTimeout = 30,       batchSize = bs,       polling = StandardPolling {pollInterval = 0.1},+      pollRetry = defaultPollRetryConfig,       deadLetterConfig = Nothing,       maxRetries = 3,       fifoConfig = Nothing,

@@ -4,12 +4,23 @@  import Data.Aeson (Value (..), object, (.=)) import Data.Aeson.KeyMap qualified as KeyMap+import Data.IORef (IORef, atomicModifyIORef', newIORef, readIORef) import Data.Int (Int32)-import Data.Time (NominalDiffTime)+import Data.Time (NominalDiffTime, UTCTime (..), fromGregorian)+import Data.Vector qualified as Vector+import Effectful (Eff, IOE, liftIO, runEff)+import Effectful.Dispatch.Dynamic (interpret)+import Effectful.Error.Static (Error, runErrorNoCallStack, throwError)+import Hasql.Errors qualified as HasqlErrors+import Pgmq.Effectful (Pgmq, PgmqRuntimeError (..))+import Pgmq.Effectful.Effect qualified as PgmqEffect import Pgmq.Hasql.Statements.Types (ReadGrouped (..), ReadMessage (..), ReadWithPollMessage (..))-import Pgmq.Types (parseQueueName)+import Pgmq.Types (Message (..), MessageBody (..), MessageId (..), parseQueueName) import Shibuya.Adapter.Pgmq.Config   ( PgmqAdapterConfig (..),+    PollRetryConfig (..),+    PollingConfig (..),+    defaultPollRetryConfig,     defaultPollingConfig,   ) import Shibuya.Adapter.Pgmq.Internal@@ -18,7 +29,9 @@     mkReadMessage,     mkReadWithPoll,     nominalToSeconds,+    pgmqChunks,   )+import Streamly.Data.Stream qualified as Stream import Test.Hspec  spec :: Spec@@ -27,6 +40,7 @@   mkReadMessageSpec   mkReadWithPollSpec   mkReadGroupedSpec+  pollRetrySpec   mergeDlqHeadersSpec  -- | Tests for nominalToSeconds@@ -74,6 +88,7 @@             visibilityTimeout = 60,             batchSize = 10,             polling = defaultPollingConfig,+            pollRetry = defaultPollRetryConfig,             deadLetterConfig = Nothing,             maxRetries = 3,             fifoConfig = Nothing,@@ -110,6 +125,7 @@             visibilityTimeout = 60,             batchSize = 10,             polling = defaultPollingConfig,+            pollRetry = defaultPollRetryConfig,             deadLetterConfig = Nothing,             maxRetries = 3,             fifoConfig = Nothing,@@ -154,6 +170,7 @@             visibilityTimeout = 60,             batchSize = 20,             polling = defaultPollingConfig,+            pollRetry = defaultPollRetryConfig,             deadLetterConfig = Nothing,             maxRetries = 3,             fifoConfig = Nothing,@@ -174,6 +191,35 @@   it "sets qty to batchSize" $ do     queryQty `shouldBe` 20 +pollRetrySpec :: Spec+pollRetrySpec = describe "pgmqChunks poll retry" $ do+  it "retries transient poll errors and returns the successful batch" $ do+    calls <- newIORef (0 :: Int)+    let cfg = retryTestConfig 5+    result <-+      runStubPgmq calls (transientThenSuccess 2) $+        Stream.toList (Stream.take 1 (pgmqChunks cfg))+    result `shouldBe` Right [Vector.singleton testMessage]+    readIORef calls `shouldReturn` 3++  it "does not retry permanent poll errors" $ do+    calls <- newIORef (0 :: Int)+    let cfg = retryTestConfig 5+    result <-+      runStubPgmq calls (const (Left permanentError)) $+        Stream.toList (Stream.take 1 (pgmqChunks cfg))+    result `shouldBe` Left permanentError+    readIORef calls `shouldReturn` 1++  it "stops retrying after maxAttempts is exhausted" $ do+    calls <- newIORef (0 :: Int)+    let cfg = retryTestConfig 2+    result <-+      runStubPgmq calls (const (Left transientError)) $+        Stream.toList (Stream.take 1 (pgmqChunks cfg))+    result `shouldBe` Left transientError+    readIORef calls `shouldReturn` 2+ -- | Tests for mergeDlqHeaders, the helper that injects the failing -- consumer's trace context onto a DLQ message while preserving the -- original producer's trace under x-shibuya-upstream-* keys.@@ -253,3 +299,78 @@         KeyMap.lookup "x-shibuya-upstream-tracestate" obj           `shouldBe` Just (String "vendor=opaque")       other -> expectationFailure $ "expected merged Object, got " <> show other++retryTestConfig :: Int -> PgmqAdapterConfig+retryTestConfig attempts =+  let queueName = case parseQueueName "retry_test" of+        Right q -> q+        Left e -> error $ "Unexpected: " <> show e+   in PgmqAdapterConfig+        { queueName = queueName,+          visibilityTimeout = 30,+          batchSize = 1,+          polling = StandardPolling {pollInterval = 0},+          pollRetry =+            PollRetryConfig+              { maxAttempts = attempts,+                initialBackoff = 0,+                maxBackoff = 0+              },+          deadLetterConfig = Nothing,+          maxRetries = 3,+          fifoConfig = Nothing,+          prefetchConfig = Nothing+        }++runStubPgmq ::+  IORef Int ->+  (Int -> Either PgmqRuntimeError (Vector.Vector Message)) ->+  Eff '[Pgmq, Error PgmqRuntimeError, IOE] a ->+  IO (Either PgmqRuntimeError a)+runStubPgmq calls respond action =+  runEff $+    runErrorNoCallStack $+      interpret+        ( \_ -> \case+            PgmqEffect.ReadMessage _ -> nextPoll+            PgmqEffect.ReadWithPoll _ -> nextPoll+            PgmqEffect.ReadGrouped _ -> nextPoll+            PgmqEffect.ReadGroupedWithPoll _ -> nextPoll+            PgmqEffect.ReadGroupedRoundRobin _ -> nextPoll+            PgmqEffect.ReadGroupedRoundRobinWithPoll _ -> nextPoll+            _ -> error "unexpected Pgmq operation in retry test"+        )+        action+  where+    nextPoll = do+      n <- liftIO $ atomicModifyIORef' calls (\c -> let c' = c + 1 in (c', c'))+      case respond n of+        Left err -> throwError err+        Right batch -> pure batch++transientThenSuccess :: Int -> Int -> Either PgmqRuntimeError (Vector.Vector Message)+transientThenSuccess failures n+  | n <= failures = Left transientError+  | otherwise = Right (Vector.singleton testMessage)++transientError :: PgmqRuntimeError+transientError = PgmqAcquisitionTimeout++permanentError :: PgmqRuntimeError+permanentError =+  PgmqConnectionError (HasqlErrors.AuthenticationConnectionError "bad password")++testMessage :: Message+testMessage =+  Message+    { messageId = MessageId 1,+      visibilityTime = testTime,+      enqueuedAt = testTime,+      lastReadAt = Nothing,+      readCount = 1,+      body = MessageBody Null,+      headers = Nothing+    }++testTime :: UTCTime+testTime = UTCTime (fromGregorian 2026 1 1) 0