shibuya-pgmq-adapter 0.3.0.0 → 0.4.0.0

raw patch · 7 files changed

+122/−7 lines, 7 filesPVP ok

version bump matches the API change (PVP)

API changes (from Hackage documentation)

Files

• CHANGELOG.md +23/−0
• shibuya-pgmq-adapter.cabal +3/−3
• src/Shibuya/Adapter/Pgmq.hs +18/−0
• src/Shibuya/Adapter/Pgmq/Convert.hs +13/−1
• src/Shibuya/Adapter/Pgmq/Internal.hs +19/−2
• test/Shibuya/Adapter/Pgmq/ConvertSpec.hs +29/−1
• test/Shibuya/Adapter/Pgmq/InternalSpec.hs +17/−0

@@ -1,5 +1,28 @@ # Changelog +## 0.4.0.0 — 2026-04-29++Paired with `shibuya-core 0.4.0.0`.++### Additions++- Envelopes now carry the delivery `attempt` counter (from pgmq's+  `readCount`, zero-indexed), enabling exponential backoff via+  `Shibuya.Core.Retry`. The first delivery sees `Just (Attempt 0)`, the+  first retry `Just (Attempt 1)`, and so on.++### Internal++- `nominalToSeconds` (in `Shibuya.Adapter.Pgmq.Internal`) now clamps to+  the `Int32` range instead of silently wrapping. Misconfigured+  retry/lease durations cap at ~68 years rather than producing+  undefined behavior on the visibility-timeout offset passed to pgmq.++### Compatibility++- Requires `shibuya-core ^>=0.4.0.0` for the `Attempt` type and the+  `attempt` field on `Envelope`.+ ## 0.3.0.0 — 2026-04-24  Upgraded to `pgmq-hs` 0.2.0.0 series

@@ -1,6 +1,6 @@ cabal-version: 3.14 name: shibuya-pgmq-adapter-version: 0.3.0.0+version: 0.4.0.0 synopsis: PGMQ adapter for the Shibuya queue processing framework description:   A Shibuya adapter that integrates with pgmq (PostgreSQL Message Queue)@@ -46,7 +46,7 @@     pgmq-core ^>=0.2,     pgmq-effectful ^>=0.2,     pgmq-hasql ^>=0.2,-    shibuya-core ^>=0.3.0.0,+    shibuya-core ^>=0.4.0.0,     stm ^>=2.5,     streamly ^>=0.11,     streamly-core ^>=0.3,@@ -113,7 +113,7 @@     pgmq-migration ^>=0.2,     quickcheck-instances ^>=0.3,     random,-    shibuya-core ^>=0.3.0.0,+    shibuya-core ^>=0.4.0.0,     shibuya-pgmq-adapter,     stm,     streamly ^>=0.11,

@@ -43,6 +43,24 @@ -- 'readCount' exceeds 'maxRetries' in the config, it is automatically -- dead-lettered before being passed to the handler. --+-- The current attempt (zero-indexed) is exposed on the envelope as+-- @envelope.attempt :: Maybe Attempt@. Handlers wanting exponential+-- backoff can use 'Shibuya.Core.Retry.retryWithBackoff':+--+-- @+-- import Shibuya.Core.Retry (defaultBackoffPolicy, retryWithBackoff)+--+-- handler ingested = do+--   result <- tryProcess ingested.envelope.payload+--   case result of+--     Right () -> pure AckOk+--     Left _  -> retryWithBackoff defaultBackoffPolicy ingested.envelope+-- @+--+-- The adapter clamps any 'Shibuya.Core.Ack.RetryDelay' to fit within+-- pgmq's @Int32@ second range (~68 years), so a long+-- 'Shibuya.Core.Retry.BackoffPolicy.maxDelay' is safe.+-- -- == FIFO Support -- -- For ordered message processing, configure 'fifoConfig'. Messages are

@@ -25,7 +25,7 @@ import Data.Text.Encoding qualified as TE import Pgmq.Types qualified as Pgmq import Shibuya.Core.Ack (DeadLetterReason (..))-import Shibuya.Core.Types (Cursor (..), Envelope (..), MessageId (..), TraceHeaders)+import Shibuya.Core.Types (Attempt (..), Cursor (..), Envelope (..), MessageId (..), TraceHeaders)  -- | Convert a pgmq MessageId to a Shibuya MessageId. -- pgmq uses Int64, Shibuya uses Text.@@ -80,6 +80,7 @@ -- | Convert a pgmq Message to a Shibuya Envelope. -- The payload is the raw JSON Value from pgmq. -- Extracts W3C trace context from headers if present.+-- Populates the delivery 'attempt' counter from pgmq's 'readCount'. pgmqMessageToEnvelope :: Pgmq.Message -> Envelope Value pgmqMessageToEnvelope msg =   Envelope@@ -88,8 +89,19 @@       partition = extractPartition msg.headers,       enqueuedAt = Just msg.enqueuedAt,       traceContext = extractTraceHeaders msg.headers,+      attempt = Just (readCountToAttempt msg.readCount),       payload = Pgmq.unMessageBody msg.body     }++-- | Convert pgmq's @readCount@ (1-based, incremented on read) to a Shibuya+-- 'Attempt' (0-based delivery counter).+--+-- pgmq increments @readCount@ before exposing the message, so on the first+-- delivery @readCount = 1@ which corresponds to @Attempt 0@. The @max 0@+-- clamp guards the unexpected @readCount = 0@ case (which pgmq does not+-- emit but cannot be ruled out at the type level).+readCountToAttempt :: Int64 -> Attempt+readCountToAttempt rc = Attempt (fromIntegral (max 0 (rc - 1)))  -- | Create a dead-letter queue payload with optional metadata. mkDlqPayload ::

@@ -90,9 +90,26 @@ import Streamly.Data.Stream.Prelude qualified as StreamP import Streamly.Data.Unfold qualified as Unfold --- | Convert NominalDiffTime to seconds as Int32 for pgmq.+-- | Convert 'NominalDiffTime' to seconds as 'Int32', saturating at the+-- 'Int32' bounds.+--+-- Used when extending pgmq visibility timeouts ('AckRetry', 'AckHalt', and+-- lease extension). pgmq's @changeVisibilityTimeout@ accepts 'Int32'+-- seconds; values larger than @maxBound@ (~68 years) silently wrap under+-- the previous @ceiling . nominalDiffTimeToSeconds@ implementation. This+-- helper saturates instead, so a misconfigured 'BackoffPolicy.maxDelay'+-- produces a merely-very-long retry rather than a corrupt or+-- panic-inducing one. nominalToSeconds :: NominalDiffTime -> Int32-nominalToSeconds = ceiling . nominalDiffTimeToSeconds+nominalToSeconds dt =+  let seconds :: Double+      seconds = realToFrac (nominalDiffTimeToSeconds dt)+      maxSec :: Double+      maxSec = fromIntegral (maxBound :: Int32)+      minSec :: Double+      minSec = fromIntegral (minBound :: Int32)+      clamped = max minSec (min maxSec seconds)+   in ceiling clamped  -- | Create a ReadMessage query from config. mkReadMessage :: PgmqAdapterConfig -> ReadMessage

@@ -8,7 +8,7 @@ import Pgmq.Types qualified as Pgmq import Shibuya.Adapter.Pgmq.Convert import Shibuya.Core.Ack (DeadLetterReason (..))-import Shibuya.Core.Types (Cursor (..), Envelope (..), MessageId (..))+import Shibuya.Core.Types (Attempt (..), Cursor (..), Envelope (..), MessageId (..)) import Test.Hspec import Test.QuickCheck @@ -271,6 +271,34 @@         msg = mkMessage 42 (String "test") hdrs         Envelope {traceContext = envTraceContext} = pgmqMessageToEnvelope msg     envTraceContext `shouldBe` Nothing++  describe "attempt field" $ do+    let mkMessageWithReadCount rc =+          Pgmq.Message+            { messageId = Pgmq.MessageId 42,+              visibilityTime = sampleTime,+              enqueuedAt = sampleTime,+              lastReadAt = Nothing,+              readCount = rc,+              body = Pgmq.MessageBody (String "test"),+              headers = Nothing+            }++    it "first delivery (readCount=1) -> Attempt 0" $ do+      let env = pgmqMessageToEnvelope (mkMessageWithReadCount 1)+      env.attempt `shouldBe` Just (Attempt 0)++    it "first retry (readCount=2) -> Attempt 1" $ do+      let env = pgmqMessageToEnvelope (mkMessageWithReadCount 2)+      env.attempt `shouldBe` Just (Attempt 1)++    it "fifth read (readCount=5) -> Attempt 4" $ do+      let env = pgmqMessageToEnvelope (mkMessageWithReadCount 5)+      env.attempt `shouldBe` Just (Attempt 4)++    it "boundary readCount=0 clamps to Attempt 0" $ do+      let env = pgmqMessageToEnvelope (mkMessageWithReadCount 0)+      env.attempt `shouldBe` Just (Attempt 0)  -- | Tests for mkDlqPayload mkDlqPayloadSpec :: Spec

@@ -1,5 +1,7 @@ module Shibuya.Adapter.Pgmq.InternalSpec (spec) where +import Data.Int (Int32)+import Data.Time (NominalDiffTime) import Pgmq.Hasql.Statements.Types (ReadGrouped (..), ReadMessage (..), ReadWithPollMessage (..)) import Pgmq.Types (parseQueueName) import Shibuya.Adapter.Pgmq.Config@@ -38,6 +40,21 @@    it "handles negative (rounds toward positive infinity)" $ do     nominalToSeconds (-5.1) `shouldBe` (-5)++  describe "clamping" $ do+    it "exact 30 seconds passes through" $ do+      nominalToSeconds 30 `shouldBe` 30++    it "rounds up subsecond" $ do+      nominalToSeconds 0.1 `shouldBe` 1++    it "clamps a 100-year delay to maxBound :: Int32" $ do+      let hundredYears = 100 * 365 * 24 * 60 * 60 :: NominalDiffTime+      nominalToSeconds hundredYears `shouldBe` (maxBound :: Int32)++    it "clamps a very large negative delay to minBound :: Int32" $ do+      let hugelyNegative = negate (100 * 365 * 24 * 60 * 60) :: NominalDiffTime+      nominalToSeconds hugelyNegative `shouldBe` (minBound :: Int32)  -- | Tests for mkReadMessage mkReadMessageSpec :: Spec