diff --git a/CHANGELOG.md b/CHANGELOG.md --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -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`. diff --git a/shibuya-pgmq-adapter.cabal b/shibuya-pgmq-adapter.cabal --- a/shibuya-pgmq-adapter.cabal +++ b/shibuya-pgmq-adapter.cabal @@ -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) diff --git a/src/Shibuya/Adapter/Pgmq.hs b/src/Shibuya/Adapter/Pgmq.hs --- a/src/Shibuya/Adapter/Pgmq.hs +++ b/src/Shibuya/Adapter/Pgmq.hs @@ -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 diff --git a/src/Shibuya/Adapter/Pgmq/Config.hs b/src/Shibuya/Adapter/Pgmq/Config.hs --- a/src/Shibuya/Adapter/Pgmq/Config.hs +++ b/src/Shibuya/Adapter/Pgmq/Config.hs @@ -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, diff --git a/src/Shibuya/Adapter/Pgmq/Internal.hs b/src/Shibuya/Adapter/Pgmq/Internal.hs --- a/src/Shibuya/Adapter/Pgmq/Internal.hs +++ b/src/Shibuya/Adapter/Pgmq/Internal.hs @@ -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) diff --git a/test/Shibuya/Adapter/Pgmq/ConfigSpec.hs b/test/Shibuya/Adapter/Pgmq/ConfigSpec.hs --- a/test/Shibuya/Adapter/Pgmq/ConfigSpec.hs +++ b/test/Shibuya/Adapter/Pgmq/ConfigSpec.hs @@ -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 diff --git a/test/Shibuya/Adapter/Pgmq/IntegrationSpec.hs b/test/Shibuya/Adapter/Pgmq/IntegrationSpec.hs --- a/test/Shibuya/Adapter/Pgmq/IntegrationSpec.hs +++ b/test/Shibuya/Adapter/Pgmq/IntegrationSpec.hs @@ -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, diff --git a/test/Shibuya/Adapter/Pgmq/InternalSpec.hs b/test/Shibuya/Adapter/Pgmq/InternalSpec.hs --- a/test/Shibuya/Adapter/Pgmq/InternalSpec.hs +++ b/test/Shibuya/Adapter/Pgmq/InternalSpec.hs @@ -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