diff --git a/CHANGELOG.md b/CHANGELOG.md --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -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`, 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.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) 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 @@ -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, 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 @@ -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. 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 @@ -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) 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 @@ -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 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 @@ -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"