shibuya-pgmq-adapter 0.15.0.0 → 0.16.0.0

raw patch · 7 files changed

+233/−10 lines, 7 filesPVP ok

version bump matches the API change (PVP)

API changes (from Hackage documentation)

Files

• CHANGELOG.md +18/−0
• shibuya-pgmq-adapter.cabal +1/−1
• src/Shibuya/Adapter/Pgmq.hs +2/−1
• src/Shibuya/Adapter/Pgmq/Config.hs +6/−0
• src/Shibuya/Adapter/Pgmq/Internal.hs +5/−0
• test/Shibuya/Adapter/Pgmq/IntegrationSpec.hs +94/−4
• test/Shibuya/Adapter/Pgmq/InternalSpec.hs +107/−4

@@ -1,5 +1,23 @@ # Changelog +## 0.16.0.0 — 2026-09-16++### Breaking Changes++- `FifoReadStrategy` gains the public `HeadPerGroup` constructor. Exhaustive+  matches over the strategy type must handle the new case.++### Features++- `HeadPerGroup` uses PGMQ's grouped-head reads with either polling mode. A+  failed or delayed head blocks only its own group while a batch can still+  contain heads from many groups. PGMQ 1.12.0 or later is required.++### Tests and Benchmarks++- Added exact dispatch coverage, PostgreSQL failure/delay integration tests,+  and a full-drain performance matrix for safe FIFO batch sizes 1, 10, and 50.+ ## 0.15.0.0 — 2026-09-14  Driven by the `pgmq-hs` 0.6 release. The adapter remains paired with `shibuya-core 0.9.0.0`,

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

@@ -64,10 +64,11 @@ -- == FIFO Support -- -- For ordered message processing, configure 'fifoConfig'. Messages are--- grouped by the @x-pgmq-group@ header. Two strategies are available:+-- grouped by the @x-pgmq-group@ header. Three strategies are available: -- -- * 'ThroughputOptimized': Fill batches from the same group (SQS-like) -- * 'RoundRobin': Fair distribution across groups+-- * 'HeadPerGroup': Lease at most one absolute head from each group (PGMQ 1.12+) module Shibuya.Adapter.Pgmq   ( -- * Adapter     pgmqAdapter,

@@ -256,6 +256,12 @@     ThroughputOptimized   | -- | Fair round-robin distribution across groups.     RoundRobin+  | -- | Lease at most one absolute head from each group (PGMQ 1.12+).+    --+    -- An invisible or delayed head blocks its group. 'PgmqAdapterConfig.batchSize'+    -- bounds the number of groups claimed by one read, rather than the number of+    -- members claimed from a single group.+    HeadPerGroup   deriving stock (Show, Eq, Generic)  -- | Default polling configuration using standard polling with 1 second interval.

@@ -67,6 +67,8 @@     changeVisibilityTimeout,     deleteMessage,     readGrouped,+    readGroupedHead,+    readGroupedHeadWithPoll,     readGroupedRoundRobin,     readGroupedRoundRobinWithPoll,     readGroupedWithPoll,@@ -466,6 +468,7 @@         result <- case fifo.readStrategy of           ThroughputOptimized -> readGrouped (mkReadGrouped config)           RoundRobin -> readGroupedRoundRobin (mkReadGrouped config)+          HeadPerGroup -> readGroupedHead (mkReadGrouped config)         when (Vector.null result) $           liftIO $             threadDelay (nominalToMicros interval)@@ -476,6 +479,8 @@             readGroupedWithPoll (mkReadGroupedWithPoll config maxSec intervalMs)           RoundRobin ->             readGroupedRoundRobinWithPoll (mkReadGroupedWithPoll config maxSec intervalMs)+          HeadPerGroup ->+            readGroupedHeadWithPoll (mkReadGroupedWithPoll config maxSec intervalMs)      nominalToMicros :: NominalDiffTime -> Int     nominalToMicros t = floor (nominalDiffTimeToSeconds t * 1_000_000)

@@ -14,25 +14,39 @@ import Data.Aeson (Value (..), object, (.=)) import Data.IORef (newIORef, readIORef, writeIORef) import Data.Int (Int32)+import Data.List (sort)+import Data.Text (Text)+import Data.Time (addUTCTime, getCurrentTime) import Data.Vector qualified as Vector import Effectful (Eff, IOE, liftIO, runEff) import Effectful.Error.Static (Error, runErrorNoCallStack) import Hasql.Pool qualified as Pool+import Hasql.Session qualified as Hasql import Pgmq.Effectful (Pgmq, PgmqRuntimeError, runPgmq) import Pgmq.Effectful qualified as PgmqEff import Pgmq.Hasql.Sessions qualified as Sessions-import Pgmq.Hasql.Statements.Types (MessageQuery (..), ReadMessage (..), SendMessage (..), VisibilityTimeoutQuery (..))-import Pgmq.Types (MessageBody (..), QueueName)+import Pgmq.Hasql.Statements.Types+  ( MessageQuery (..),+    ReadMessage (..),+    SendMessage (..),+    SendMessageWithHeaders (..),+    SendMessageWithHeadersForLater (..),+    VisibilityTimeoutQuery (..),+  )+import Pgmq.Types (Message (..), MessageBody (..), MessageHeaders (..), MessageId, QueueName) import Shibuya.Adapter.Pgmq.Config-  ( PgmqAdapterConfig (..),+  ( FifoConfig (..),+    FifoReadStrategy (..),+    PgmqAdapterConfig (..),     PollingConfig (..),     defaultConfig,     defaultPollRetryConfig,   ) import Shibuya.Adapter.Pgmq.Convert (pgmqMessageToEnvelope)-import Shibuya.Adapter.Pgmq.Internal (mkLease)+import Shibuya.Adapter.Pgmq.Internal (mkLease, pgmqChunks) import Shibuya.Core.Lease (Lease (..)) import Shibuya.Core.Types (Envelope (..))+import Streamly.Data.Stream qualified as Stream import System.Environment (lookupEnv) import Test.Hspec import TmpPostgres (TestFixture (..), runPgmqSession, withPgmqDb, withTestFixture)@@ -51,6 +65,7 @@       basicMessageProcessingSpec       visibilityTimeoutSpec       retryHandlingSpec+      groupedHeadSpec  -- | Wrapper to run tests with a temporary database and fixture withTempDbFixture :: (TestFixture -> IO ()) -> IO ()@@ -343,6 +358,81 @@             }       pure $ Vector.length msgs     count2 `shouldBe` 1++groupedHeadSpec :: SpecWith TestFixture+groupedHeadSpec = describe "Grouped-head FIFO reads" $ do+  it "leases at most one absolute head per group and advances only a settled group" $ \TestFixture {pool, queueName, dlqName = _} -> do+    runPgmqSession pool $ Sessions.createFifoIndex queueName+    (a1, a2, b1) <-+      runPgmqSession pool $ do+        a1 <- sendGrouped queueName "a" "a1"+        a2 <- sendGrouped queueName "a" "a2"+        b1 <- sendGrouped queueName "b" "b1"+        _ <- sendGrouped queueName "b" "b2"+        pure (a1, a2, b1)++    heads <- readGroupedHeadBatch pool queueName+    sortedMessageIds heads `shouldBe` sort [a1, b1]++    blocked <- readGroupedHeadBatch pool queueName+    blocked `shouldBe` Vector.empty++    deleted <- runAdapterIO pool (PgmqEff.deleteMessage (MessageQuery queueName a1))+    deleted `shouldBe` True+    advanced <- readGroupedHeadBatch pool queueName+    sortedMessageIds advanced `shouldBe` [a2]++  it "lets another group advance while a delayed absolute head blocks its successor" $ \TestFixture {pool, queueName, dlqName = _} -> do+    runPgmqSession pool $ Sessions.createFifoIndex queueName+    scheduledAt <- addUTCTime 60 <$> getCurrentTime+    _a1 <-+      runPgmqSession pool $+        Sessions.sendMessageWithHeadersForLater+          SendMessageWithHeadersForLater+            { queueName,+              messageBody = MessageBody (String "a1"),+              messageHeaders = groupHeaders "a",+              scheduledAt+            }+    _a2 <- runPgmqSession pool (sendGrouped queueName "a" "a2")+    b1 <- runPgmqSession pool (sendGrouped queueName "b" "b1")++    heads <- readGroupedHeadBatch pool queueName+    sortedMessageIds heads `shouldBe` [b1]++readGroupedHeadBatch :: Pool.Pool -> QueueName -> IO (Vector.Vector Message)+readGroupedHeadBatch pool queueName = do+  batches <-+    runAdapterIO pool $+      Stream.toList $+        Stream.take 1 $+          pgmqChunks+            ( (defaultConfig queueName)+                { visibilityTimeout = 60,+                  batchSize = 10,+                  polling = StandardPolling {pollInterval = 0.01},+                  fifoConfig = Just FifoConfig {readStrategy = HeadPerGroup}+                }+            )+  case batches of+    [batch] -> pure batch+    _ -> error "Expected exactly one grouped-head batch"++sendGrouped :: QueueName -> Text -> Text -> Hasql.Session MessageId+sendGrouped queueName group body =+  Sessions.sendMessageWithHeaders+    SendMessageWithHeaders+      { queueName,+        messageBody = MessageBody (String body),+        messageHeaders = groupHeaders group,+        delay = Just 0+      }++groupHeaders :: Text -> MessageHeaders+groupHeaders group = MessageHeaders (object ["x-pgmq-group" .= group])++sortedMessageIds :: Vector.Vector Message -> [MessageId]+sortedMessageIds = sort . map (\message -> message.messageId) . Vector.toList  -- | Helper to create a config with sensible defaults for testing _mkConfig :: QueueName -> PgmqAdapterConfig

@@ -7,18 +7,21 @@ import Data.ByteString qualified as BS import Data.IORef (IORef, atomicModifyIORef', newIORef, readIORef) import Data.Int (Int32)+import Data.Text (Text) import Data.Time (NominalDiffTime, UTCTime (..), fromGregorian) import Data.Vector qualified as Vector-import Effectful (Eff, IOE, liftIO, runEff)+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 (Message (..), MessageBody (..), MessageId (..), parseQueueName)+import Pgmq.Hasql.Statements.Types (ReadGrouped (..), ReadGroupedWithPoll (..), ReadMessage (..), ReadWithPollMessage (..))+import Pgmq.Types (Message (..), MessageBody (..), MessageId (..), parseQueueName, queueNameToText) import Shibuya.Adapter.Pgmq.Config-  ( PgmqAdapterConfig (..),+  ( FifoConfig (..),+    FifoReadStrategy (..),+    PgmqAdapterConfig (..),     PollRetryConfig (..),     PollingConfig (..),     defaultPollRetryConfig,@@ -42,6 +45,7 @@   mkReadMessageSpec   mkReadWithPollSpec   mkReadGroupedSpec+  fifoDispatchSpec   pollRetrySpec   autoDeadLetterHookSpec   mergeDlqHeadersSpec@@ -200,6 +204,103 @@   it "sets qty to batchSize" $ do     queryQty `shouldBe` 20 +data FifoPollCall = FifoPollCall+  { operation :: !String,+    queue :: !Text,+    delay :: !Int32,+    quantity :: !Int32,+    pollParameters :: !(Maybe (Int32, Int32))+  }+  deriving stock (Eq, Show)++fifoDispatchSpec :: Spec+fifoDispatchSpec = describe "pgmqChunks FIFO dispatch" $ do+  let standard = StandardPolling {pollInterval = 1}+      long = LongPolling {maxPollSeconds = 5, pollIntervalMs = 100}+      cases =+        [ ("throughput standard", ThroughputOptimized, standard, "readGrouped", Nothing),+          ("round-robin standard", RoundRobin, standard, "readGroupedRoundRobin", Nothing),+          ("head-per-group standard", HeadPerGroup, standard, "readGroupedHead", Nothing),+          ("throughput long poll", ThroughputOptimized, long, "readGroupedWithPoll", Just (5, 100)),+          ("round-robin long poll", RoundRobin, long, "readGroupedRoundRobinWithPoll", Just (5, 100)),+          ("head-per-group long poll", HeadPerGroup, long, "readGroupedHeadWithPoll", Just (5, 100))+        ]+  mapM_+    ( \(label, strategy, pollingConfig, expectedOperation, expectedPoll) ->+        it ("selects " <> label) $ do+          (result, calls) <- observeFifoPoll strategy pollingConfig+          result `shouldBe` Right [Vector.singleton testMessage]+          calls+            `shouldBe` [ FifoPollCall+                           { operation = expectedOperation,+                             queue = "fifo_dispatch",+                             delay = 45,+                             quantity = 7,+                             pollParameters = expectedPoll+                           }+                       ]+    )+    cases++observeFifoPoll :: FifoReadStrategy -> PollingConfig -> IO (Either PgmqRuntimeError [Vector.Vector Message], [FifoPollCall])+observeFifoPoll strategy pollingConfig = do+  calls <- newIORef []+  result <-+    runEff $+      runErrorNoCallStack $+        interpret+          ( \_ -> \case+              PgmqEffect.ReadGrouped query -> recordStandard calls "readGrouped" query+              PgmqEffect.ReadGroupedRoundRobin query -> recordStandard calls "readGroupedRoundRobin" query+              PgmqEffect.ReadGroupedHead query -> recordStandard calls "readGroupedHead" query+              PgmqEffect.ReadGroupedWithPoll query -> recordLong calls "readGroupedWithPoll" query+              PgmqEffect.ReadGroupedRoundRobinWithPoll query -> recordLong calls "readGroupedRoundRobinWithPoll" query+              PgmqEffect.ReadGroupedHeadWithPoll query -> recordLong calls "readGroupedHeadWithPoll" query+              _ -> error "unexpected Pgmq operation in FIFO dispatch test"+          )+          (Stream.toList (Stream.take 1 (pgmqChunks (fifoDispatchConfig strategy pollingConfig))))+  observed <- readIORef calls+  pure (result, reverse observed)++recordStandard :: (IOE :> es) => IORef [FifoPollCall] -> String -> ReadGrouped -> Eff es (Vector.Vector Message)+recordStandard calls operation ReadGrouped {queueName, visibilityTimeout, qty} = do+  liftIO $+    atomicModifyIORef'+      calls+      (\xs -> (FifoPollCall operation (queueNameToText queueName) visibilityTimeout qty Nothing : xs, ()))+  pure (Vector.singleton testMessage)++recordLong :: (IOE :> es) => IORef [FifoPollCall] -> String -> ReadGroupedWithPoll -> Eff es (Vector.Vector Message)+recordLong calls operation ReadGroupedWithPoll {queueName, visibilityTimeout, qty, maxPollSeconds, pollIntervalMs} = do+  liftIO $+    atomicModifyIORef'+      calls+      ( \xs ->+          ( FifoPollCall+              operation+              (queueNameToText queueName)+              visibilityTimeout+              qty+              (Just (maxPollSeconds, pollIntervalMs))+              : xs,+            ()+          )+      )+  pure (Vector.singleton testMessage)++fifoDispatchConfig :: FifoReadStrategy -> PollingConfig -> PgmqAdapterConfig+fifoDispatchConfig strategy pollingConfig =+  let queueName = case parseQueueName "fifo_dispatch" of+        Right q -> q+        Left e -> error $ "Unexpected: " <> show e+   in (retryTestConfig 1)+        { queueName = queueName,+          visibilityTimeout = 45,+          batchSize = 7,+          polling = pollingConfig,+          fifoConfig = Just FifoConfig {readStrategy = strategy}+        }+ pollRetrySpec :: Spec pollRetrySpec = describe "pgmqChunks poll retry" $ do   it "retries transient poll errors and returns the successful batch" $ do@@ -389,6 +490,8 @@             PgmqEffect.ReadWithPoll _ -> nextPoll             PgmqEffect.ReadGrouped _ -> nextPoll             PgmqEffect.ReadGroupedWithPoll _ -> nextPoll+            PgmqEffect.ReadGroupedHead _ -> nextPoll+            PgmqEffect.ReadGroupedHeadWithPoll _ -> nextPoll             PgmqEffect.ReadGroupedRoundRobin _ -> nextPoll             PgmqEffect.ReadGroupedRoundRobinWithPoll _ -> nextPoll             _ -> error "unexpected Pgmq operation in retry test"