shibuya-pgmq-adapter-0.11.0.0: test/Shibuya/Adapter/Pgmq/ChaosSpec.hs
-- | Chaos tests for the PGMQ adapter.
--
-- These tests verify system behavior under adverse conditions:
-- - Poison messages (handler failures)
-- - Long-running handlers (lease extension)
-- - Graceful shutdown during processing
-- - Database connection issues
--
-- Run with: cabal test --test-option='-p' --test-option='/Chaos/'
module Shibuya.Adapter.Pgmq.ChaosSpec (spec) where
import Control.Concurrent (threadDelay)
import Control.Concurrent.Async (async, cancel)
import Control.Monad (forM_)
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.Text qualified as Text
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 Pgmq.Effectful (Pgmq, PgmqRuntimeError, runPgmq)
import Pgmq.Effectful qualified as PgmqEff
import Pgmq.Hasql.Sessions qualified as Sessions
import Pgmq.Hasql.Statements.Types (ReadMessage (..), SendMessage (..), SendMessageWithHeaders (..))
import Pgmq.Types (MessageBody (..), MessageHeaders (..), QueueName)
import Shibuya.Adapter (Adapter)
import Shibuya.Adapter.Pgmq
( PgmqAdapterConfig (..),
PollingConfig (..),
PrefetchConfig (..),
defaultConfig,
defaultPrefetchConfig,
directDeadLetter,
mkPgmqAdapterEnv,
pgmqAdapter,
)
import Shibuya.Adapter.Pgmq.Internal (mkIngested)
import Shibuya.App
( AppConfig (..),
ProcessorId (..),
ShutdownConfig (..),
SupervisionStrategy (..),
defaultAppConfig,
mkProcessor,
runApp,
stopAppGracefully,
)
import Shibuya.Core.Ack (AckDecision (..), DeadLetterReason (..), HaltReason (..))
import Shibuya.Core.AckHandle (AckHandle (..))
import Shibuya.Core.Ingested (Ingested (..))
import Shibuya.Handler (Handler)
import Shibuya.Telemetry.Effect (Tracing, runTracingNoop)
import System.Environment (lookupEnv)
import System.Timeout (timeout)
import Test.Hspec
import TmpPostgres (TestFixture (..), runPgmqSession, withPgmqDb, withTestFixture)
spec :: Spec
spec = do
around withTempDbFixture $ do
describe "Chaos" $ do
poisonMessageSpec
longHandlerSpec
gracefulShutdownSpec
prefetchSpec
-- | Wrapper to run tests with a temporary database and fixture
withTempDbFixture :: (TestFixture -> IO ()) -> IO ()
withTempDbFixture action = do
skipDb <- lookupEnv "PGMQ_TEST_SKIP_DB"
case skipDb of
Just "1" -> pendingWith "Database tests skipped (PGMQ_TEST_SKIP_DB=1)"
_ -> do
result <- withPgmqDb $ \pool ->
withTestFixture pool action
case result of
Left err -> expectationFailure $ "Failed to start temp database: " <> show err
Right () -> pure ()
-- | Run an Eff action with Pgmq effect against a pool, throwing on error.
runAdapterIO :: Pool.Pool -> Eff '[Pgmq, Error PgmqRuntimeError, IOE] a -> IO a
runAdapterIO pool action = do
result <- runEff $ runErrorNoCallStack $ runPgmq pool action
case result of
Left err -> error $ "Pgmq error: " <> show err
Right a -> pure a
--------------------------------------------------------------------------------
-- Poison Message Tests
--------------------------------------------------------------------------------
-- | Tests for poison message handling (handler failures).
--
-- When a handler throws an exception or returns AckDeadLetter,
-- messages should go to the DLQ (if configured).
poisonMessageSpec :: SpecWith TestFixture
poisonMessageSpec = describe "Poison messages" $ do
it "handler exception causes message to be retried" $ \TestFixture {pool, queueName, dlqName = _} -> do
-- Send a message
runPgmqSession pool $ do
_ <-
Sessions.sendMessage $
SendMessage
{ queueName = queueName,
messageBody = MessageBody (String "poison"),
delay = Just 0
}
pure ()
-- Read the message with short VT, simulate failure by not acking
runAdapterIO pool $ do
msgs <-
PgmqEff.readMessage $
ReadMessage
{ queueName = queueName,
delay = 1, -- 1 second visibility timeout
batchSize = Just 1,
conditional = Nothing
}
liftIO $ Vector.length msgs `shouldBe` 1
-- Don't delete - simulates handler failure
-- Wait for VT to expire
threadDelay 1500000
-- Message should be visible again (retry)
count <- runAdapterIO pool $ do
msgs <-
PgmqEff.readMessage $
ReadMessage
{ queueName = queueName,
delay = 30,
batchSize = Just 1,
conditional = Nothing
}
pure $ Vector.length msgs
count `shouldBe` 1
it "AckDeadLetter sends message to DLQ when configured" $ \TestFixture {pool, queueName, dlqName} -> do
-- Send a message
runPgmqSession pool $ do
_ <-
Sessions.sendMessage $
SendMessage
{ queueName = queueName,
messageBody = MessageBody (String "to-dlq"),
delay = Just 0
}
pure ()
processedRef <- newIORef (0 :: Int)
-- Create adapter with DLQ configured
let config =
(defaultConfig queueName)
{ visibilityTimeout = 5,
batchSize = 1,
polling = StandardPolling {pollInterval = 0.1},
deadLetterConfig = Just $ directDeadLetter dlqName True
}
-- Run processor that dead-letters the message
runAdapterIO pool $ runTracingNoop $ do
adapter <- requireAdapter pool config
let handler = deadLetterHandler processedRef
processor = mkProcessor adapter handler
result <- runApp defaultAppConfig [(ProcessorId "dlq-test", processor)]
case result of
Left err -> liftIO $ expectationFailure $ "Failed to start app: " <> show err
Right appHandle -> do
-- Wait for message to be processed
liftIO $ waitForProcessed processedRef 1 3000000
-- Stop the app
let shutdownConfig = ShutdownConfig {drainTimeout = 5}
_ <- stopAppGracefully shutdownConfig appHandle
pure ()
-- Verify message is in DLQ
dlqCount <-
runPgmqSession pool $
Sessions.readMessage $
ReadMessage
{ queueName = dlqName,
delay = 30,
batchSize = Just 10,
conditional = Nothing
}
Vector.length dlqCount `shouldBe` 1
-- Verify main queue is empty
mainCount <-
runPgmqSession pool $
Sessions.readMessage $
ReadMessage
{ queueName = queueName,
delay = 30,
batchSize = Just 10,
conditional = Nothing
}
Vector.length mainCount `shouldBe` 0
it "preserves trace headers when moving to DLQ" $ \TestFixture {pool, queueName, dlqName} -> do
-- Send a message with trace headers
let traceparent = "00-0af7651916cd43dd8448eb211c80319c-b7ad6b7169203331-01" :: Text.Text
tracestate = "vendor=opaque" :: Text.Text
traceHeaders = object ["traceparent" .= traceparent, "tracestate" .= tracestate]
runPgmqSession pool $ do
_ <-
Sessions.sendMessageWithHeaders $
SendMessageWithHeaders
{ queueName = queueName,
messageBody = MessageBody (String "trace-to-dlq"),
messageHeaders = MessageHeaders traceHeaders,
delay = Just 0
}
pure ()
processedRef <- newIORef (0 :: Int)
-- Create adapter with DLQ configured
let config =
(defaultConfig queueName)
{ visibilityTimeout = 5,
batchSize = 1,
polling = StandardPolling {pollInterval = 0.1},
deadLetterConfig = Just $ directDeadLetter dlqName True
}
-- Run processor that dead-letters the message
runAdapterIO pool $ runTracingNoop $ do
adapter <- requireAdapter pool config
let handler = deadLetterHandler processedRef
processor = mkProcessor adapter handler
result <- runApp defaultAppConfig [(ProcessorId "dlq-trace-test", processor)]
case result of
Left err -> liftIO $ expectationFailure $ "Failed to start app: " <> show err
Right appHandle -> do
-- Wait for message to be processed
liftIO $ waitForProcessed processedRef 1 3000000
-- Stop the app
let shutdownConfig = ShutdownConfig {drainTimeout = 5}
_ <- stopAppGracefully shutdownConfig appHandle
pure ()
-- Read the DLQ message and verify trace headers are preserved
dlqMsgs <-
runPgmqSession pool $
Sessions.readMessage $
ReadMessage
{ queueName = dlqName,
delay = 30,
batchSize = Just 10,
conditional = Nothing
}
Vector.length dlqMsgs `shouldBe` 1
let dlqMsg = Vector.head dlqMsgs
-- Verify trace headers are in the DLQ message headers
case dlqMsg.headers of
Nothing -> expectationFailure "DLQ message should have headers"
Just (Object hdrs) -> do
case KeyMap.lookup "traceparent" hdrs of
Just (String tp) -> tp `shouldBe` traceparent
_ -> expectationFailure "traceparent header should be a string"
case KeyMap.lookup "tracestate" hdrs of
Just (String ts) -> ts `shouldBe` tracestate
_ -> expectationFailure "tracestate header should be a string"
Just _ -> expectationFailure "DLQ message headers should be an object"
it "AckDeadLetter is idempotent after a successful finalize" $ \TestFixture {pool, queueName, dlqName} -> do
runPgmqSession pool $ do
_ <-
Sessions.sendMessage $
SendMessage
{ queueName = queueName,
messageBody = MessageBody (String "idempotent-dlq"),
delay = Just 0
}
pure ()
let config =
(defaultConfig queueName)
{ visibilityTimeout = 5,
batchSize = 1,
polling = StandardPolling {pollInterval = 0.1},
deadLetterConfig = Just $ directDeadLetter dlqName True
}
runAdapterIO pool $ runTracingNoop $ do
msgs <-
PgmqEff.readMessage $
ReadMessage
{ queueName = queueName,
delay = 30,
batchSize = Just 1,
conditional = Nothing
}
case Vector.uncons msgs of
Nothing -> liftIO $ expectationFailure "expected one source message"
Just (msg, _) -> do
ingestedResult <- mkIngested (mkPgmqAdapterEnv pool) config msg
case ingestedResult of
Nothing -> liftIO $ expectationFailure "message should not auto-DLQ"
Just Ingested {ack = AckHandle finalize} -> do
finalize (AckDeadLetter (PoisonPill "first"))
finalize (AckDeadLetter (PoisonPill "second"))
dlqMsgs <-
runPgmqSession pool $
Sessions.readMessage $
ReadMessage
{ queueName = dlqName,
delay = 30,
batchSize = Just 10,
conditional = Nothing
}
Vector.length dlqMsgs `shouldBe` 1
sourceMsgs <-
runPgmqSession pool $
Sessions.readMessage $
ReadMessage
{ queueName = queueName,
delay = 30,
batchSize = Just 10,
conditional = Nothing
}
Vector.length sourceMsgs `shouldBe` 0
it "AckOk is idempotent after a successful finalize" $ \TestFixture {pool, queueName, dlqName = _} -> do
runPgmqSession pool $ do
_ <-
Sessions.sendMessage $
SendMessage
{ queueName = queueName,
messageBody = MessageBody (String "idempotent-ok"),
delay = Just 0
}
pure ()
let config = (defaultConfig queueName) {visibilityTimeout = 5, batchSize = 1}
runAdapterIO pool $ runTracingNoop $ do
msgs <-
PgmqEff.readMessage $
ReadMessage
{ queueName = queueName,
delay = 30,
batchSize = Just 1,
conditional = Nothing
}
case Vector.uncons msgs of
Nothing -> liftIO $ expectationFailure "expected one source message"
Just (msg, _) -> do
ingestedResult <- mkIngested (mkPgmqAdapterEnv pool) config msg
case ingestedResult of
Nothing -> liftIO $ expectationFailure "message should not auto-DLQ"
Just Ingested {ack = AckHandle finalize} -> do
finalize AckOk
finalize AckOk
sourceMsgs <-
runPgmqSession pool $
Sessions.readMessage $
ReadMessage
{ queueName = queueName,
delay = 30,
batchSize = Just 10,
conditional = Nothing
}
Vector.length sourceMsgs `shouldBe` 0
it "AckHalt uses haltVisibilityTimeout when configured" $ \TestFixture {pool, queueName, dlqName = _} -> do
runPgmqSession pool $ do
_ <-
Sessions.sendMessage $
SendMessage
{ queueName = queueName,
messageBody = MessageBody (String "halt-vt"),
delay = Just 0
}
pure ()
let config =
(defaultConfig queueName)
{ visibilityTimeout = 30,
haltVisibilityTimeout = Just 1
}
runAdapterIO pool $ runTracingNoop $ do
msgs <-
PgmqEff.readMessage $
ReadMessage
{ queueName = queueName,
delay = 30,
batchSize = Just 1,
conditional = Nothing
}
case Vector.uncons msgs of
Nothing -> liftIO $ expectationFailure "expected one source message"
Just (msg, _) -> do
ingestedResult <- mkIngested (mkPgmqAdapterEnv pool) config msg
case ingestedResult of
Nothing -> liftIO $ expectationFailure "message should not auto-DLQ"
Just Ingested {ack = AckHandle finalize} -> finalize (AckHalt (HaltFatal "pause"))
threadDelay 1500000
visible <-
runPgmqSession pool $
Sessions.readMessage $
ReadMessage
{ queueName = queueName,
delay = 30,
batchSize = Just 1,
conditional = Nothing
}
Vector.length visible `shouldBe` 1
--------------------------------------------------------------------------------
-- Long Handler Tests
--------------------------------------------------------------------------------
-- | Tests for long-running handlers and lease extension.
longHandlerSpec :: SpecWith TestFixture
longHandlerSpec = describe "Long-running handlers" $ do
it "message stays invisible while being processed" $ \TestFixture {pool, queueName, dlqName = _} -> do
-- Send a message
runPgmqSession pool $ do
_ <-
Sessions.sendMessage $
SendMessage
{ queueName = queueName,
messageBody = MessageBody (String "long-running"),
delay = Just 0
}
pure ()
processedRef <- newIORef (0 :: Int)
-- Create adapter with longer VT
let config =
(defaultConfig queueName)
{ visibilityTimeout = 5, -- 5 second VT
batchSize = 1,
polling = StandardPolling {pollInterval = 0.1}
}
-- Start processor with slow handler
processorAsync <- async $ runAdapterIO pool $ runTracingNoop $ do
adapter <- requireAdapter pool config
let handler = slowHandler processedRef 2000000 -- 2 second delay
processor = mkProcessor adapter handler
result <- runApp defaultAppConfig [(ProcessorId "slow-test", processor)]
case result of
Left err -> liftIO $ expectationFailure $ "Failed to start app: " <> show err
Right appHandle -> do
-- Wait for message to be processed
liftIO $ waitForProcessed processedRef 1 5000000
-- Stop the app
let shutdownConfig = ShutdownConfig {drainTimeout = 5}
_ <- stopAppGracefully shutdownConfig appHandle
pure ()
-- While handler is running, try to read the message from another connection
-- It should not be visible due to VT
threadDelay 500000 -- Wait a bit for handler to pick up message
-- Try to read - should get nothing since message is being processed
count <- runPgmqSession pool $ do
msgs <-
Sessions.readMessage $
ReadMessage
{ queueName = queueName,
delay = 30,
batchSize = Just 1,
conditional = Nothing
}
pure $ Vector.length msgs
count `shouldBe` 0
-- Wait for processor to finish
_ <- cancel processorAsync
pure ()
--------------------------------------------------------------------------------
-- Graceful Shutdown Tests
--------------------------------------------------------------------------------
-- | Tests for graceful shutdown behavior.
gracefulShutdownSpec :: SpecWith TestFixture
gracefulShutdownSpec = describe "Graceful shutdown" $ do
it "processes in-flight messages during shutdown" $ \TestFixture {pool, queueName, dlqName = _} -> do
-- Send multiple messages
forM_ [1 .. 5 :: Int] $ \i ->
runPgmqSession pool $ do
_ <-
Sessions.sendMessage $
SendMessage
{ queueName = queueName,
messageBody = MessageBody (String $ Text.pack ("msg-" <> show i)),
delay = Just 0
}
pure ()
processedRef <- newIORef (0 :: Int)
let config =
(defaultConfig queueName)
{ visibilityTimeout = 30,
batchSize = 5,
polling = StandardPolling {pollInterval = 0.1}
}
-- Run processor with slow handler
runAdapterIO pool $ runTracingNoop $ do
adapter <- requireAdapter pool config
let handler = slowHandler processedRef 50000 -- 0.05 second delay per message
processor = mkProcessor adapter handler
appResult <- runApp defaultAppConfig [(ProcessorId "drain-test", processor)]
case appResult of
Left err -> liftIO $ expectationFailure $ "Failed to start app: " <> show err
Right appHandle -> do
-- Wait for all messages to be processed
liftIO $ waitForProcessed processedRef 5 5000000
-- Graceful shutdown
let shutdownConfig = ShutdownConfig {drainTimeout = 2}
_ <- stopAppGracefully shutdownConfig appHandle
pure ()
-- All messages should have been processed
processed <- readIORef processedRef
processed `shouldBe` 5
it "stops accepting new messages after shutdown signal" $ \TestFixture {pool, queueName, dlqName = _} -> do
-- Send initial messages
forM_ [1 .. 3 :: Int] $ \i ->
runPgmqSession pool $ do
_ <-
Sessions.sendMessage $
SendMessage
{ queueName = queueName,
messageBody = MessageBody (String $ Text.pack ("initial-" <> show i)),
delay = Just 0
}
pure ()
processedRef <- newIORef (0 :: Int)
let config =
(defaultConfig queueName)
{ visibilityTimeout = 30,
batchSize = 1,
polling = StandardPolling {pollInterval = 0.5} -- Slower polling
}
runAdapterIO pool $ runTracingNoop $ do
adapter <- requireAdapter pool config
let handler = countingHandler processedRef
processor = mkProcessor adapter handler
appResult <- runApp defaultAppConfig [(ProcessorId "stop-test", processor)]
case appResult of
Left err -> liftIO $ expectationFailure $ "Failed to start app: " <> show err
Right appHandle -> do
-- Wait for at least one message to be processed
liftIO $ waitForProcessed processedRef 1 3000000
-- Send more messages while still running
liftIO $
runPgmqSession pool $ do
_ <-
Sessions.sendMessage $
SendMessage
{ queueName = queueName,
messageBody = MessageBody (String "after-stop"),
delay = Just 0
}
pure ()
-- Shutdown quickly
let shutdownConfig = ShutdownConfig {drainTimeout = 1}
_ <- stopAppGracefully shutdownConfig appHandle
pure ()
-- At least some messages should remain in queue
-- (the ones sent after stop signal started)
remaining <- runPgmqSession pool $ do
msgs <-
Sessions.readMessage $
ReadMessage
{ queueName = queueName,
delay = 30,
batchSize = Just 10,
conditional = Nothing
}
pure $ Vector.length msgs
-- This test verifies the adapter stops producing after shutdown
-- Some messages may or may not be left depending on timing
remaining `shouldSatisfy` (>= 0) -- Just ensure no crash
--------------------------------------------------------------------------------
-- Prefetch Tests (scoped ConcUnlift)
--------------------------------------------------------------------------------
-- | Regression test for the historical prefetch deadlock.
--
-- With @prefetchConfig = Just ...@ the polling stage runs on a streamly
-- @parBuffered@ worker thread that must unlift @Eff@. Under effectful's default
-- 'SeqUnlift' this deadlocks ("thread blocked indefinitely in an STM
-- transaction"); the adapter now scopes the concurrent stage to 'ConcUnlift'
-- (see 'Shibuya.Adapter.Pgmq.Internal.pgmqChunksPrefetch').
--
-- Acceptance is behavioral: the queue drains to completion within a timeout. If
-- the deadlock were present the 'timeout' fires and the test fails instead of
-- hanging the suite. Stripping the @morphInner (withUnliftStrategy ...)@ wrapper
-- from 'pgmqChunksPrefetch' makes this test fail (verified during M2).
prefetchSpec :: SpecWith TestFixture
prefetchSpec = describe "Prefetch" $ do
it "drains the queue under concurrent prefetch without deadlocking" $ \TestFixture {pool, queueName, dlqName = _} -> do
let total = 50 :: Int
forM_ [1 .. total] $ \i ->
runPgmqSession pool $ do
_ <-
Sessions.sendMessage $
SendMessage
{ queueName = queueName,
messageBody = MessageBody (String $ Text.pack ("prefetch-" <> show i)),
delay = Just 0
}
pure ()
processedRef <- newIORef (0 :: Int)
let config =
(defaultConfig queueName)
{ visibilityTimeout = 30,
batchSize = 5,
polling = StandardPolling {pollInterval = 0.05},
prefetchConfig = Just defaultPrefetchConfig
}
-- Guard against a hang: a deadlock makes 'timeout' return Nothing and the
-- test fails, rather than blocking the whole suite indefinitely.
result <- timeout 30_000_000 $ runAdapterIO pool $ runTracingNoop $ do
adapter <- requireAdapter pool config
let handler = countingHandler processedRef
processor = mkProcessor adapter handler
appResult <- runApp defaultAppConfig [(ProcessorId "prefetch-test", processor)]
case appResult of
Left err -> liftIO $ expectationFailure $ "Failed to start app: " <> show err
Right appHandle -> do
liftIO $ waitForProcessed processedRef total 20_000_000
let shutdownConfig = ShutdownConfig {drainTimeout = 5}
_ <- stopAppGracefully shutdownConfig appHandle
pure ()
case result of
Nothing ->
expectationFailure
"Prefetch run timed out — likely the parBuffered/effectful STM deadlock"
Just () -> pure ()
processed <- readIORef processedRef
processed `shouldBe` total
-- All processed messages were AckOk'd, so the source queue is drained.
remaining <-
runPgmqSession pool $ do
msgs <-
Sessions.readMessage $
ReadMessage
{ queueName = queueName,
delay = 30,
batchSize = Just 100,
conditional = Nothing
}
pure $ Vector.length msgs
remaining `shouldBe` 0
-- Differential shutdown-release measurement: does prefetch hold messages
-- invisible at shutdown that the non-prefetch path would release?
--
-- 'parBuffered' buffers chunks upstream of the EP-27 shutdown gate
-- ('takeWhileM'/'releaseMessages'). On shutdown the gate releases only the
-- chunk it pulls next, so chunks still sitting in the parBuffered buffer may
-- never reach 'releaseMessages' and stay invisible until their VT expires.
--
-- The metric @held = total - processed - visibleImmediatelyAfterShutdown@
-- counts messages left invisible in the queue (inbox residual + any buffer
-- leak); it cancels out never-read and deleted messages. We run the identical
-- scenario with prefetch OFF (on the main queue) and ON (on the DLQ queue,
-- reused here as an independent second queue) and compare.
it "strands only a bounded (<= one prefetch buffer) number of extra messages invisible on shutdown" $ \TestFixture {pool, queueName, dlqName} -> do
(totalOff, procOff, visOff) <- liftIO $ measureShutdownRelease pool queueName Nothing
(totalOn, procOn, visOn) <- liftIO $ measureShutdownRelease pool dlqName (Just defaultPrefetchConfig)
let heldOff = totalOff - procOff - visOff
heldOn = totalOn - procOn - visOn
liftIO $
putStrLn $
"PREFETCH-SHUTDOWN-DIAG: off(total="
<> show totalOff
<> " processed="
<> show procOff
<> " visible="
<> show visOff
<> " held="
<> show heldOff
<> ") on(total="
<> show totalOn
<> " processed="
<> show procOn
<> " visible="
<> show visOn
<> " held="
<> show heldOn
<> ")"
-- ACCEPTED, DOCUMENTED behaviour (M3 decision, 2026-07-04): prefetch strands
-- up to bufferSize*batchSize extra messages invisible on shutdown, because
-- 'parBuffered' buffers chunks upstream of the shutdown gate and only the
-- chunk the gate pulls next is released. This is bounded and loss-free (the
-- messages are redelivered after their VT — proven by the "loses no messages
-- under prefetch" test), and is documented in the 'PrefetchConfig' Haddock
-- and the adapter ARCHITECTURE docs.
--
-- This assertion is a regression guard on the *bound*, not the exact count:
-- the messages held invisible with prefetch must not exceed one full prefetch
-- buffer (bufferSize * batchSize) plus the ordinary core-inbox residual. An
-- unbounded leak (stranding most of the queue) would fail here. We bound
-- heldOn absolutely rather than (heldOn - heldOff), because the baseline's
-- inbox residual varies run to run (heldOff has been observed at 1–3). Config
-- under test: batchSize=2, defaultPrefetchConfig bufferSize=4, runApp inbox=2.
-- Measured numbers are on the PREFETCH-SHUTDOWN-DIAG line above.
let bufferBound = fromIntegral defaultPrefetchConfig.bufferSize * 2 -- bufferSize * batchSize
inboxResidualSlack = 6 -- generous allowance for the inbox=2 + in-flight residual
heldOn `shouldSatisfy` (<= bufferBound + inboxResidualSlack)
-- Proves the "accept and document" decision is safe: the shutdown strand
-- above delays redelivery but LOSES NOTHING. A blocking handler forces
-- messages to be read ahead into the prefetch buffer without ever being
-- acked; after shutdown and one visibility-timeout window, every message is
-- recoverable (processed + still-in-queue == total). This is the empirical
-- basis for the no-data-loss claim in the docs/Haddocks.
it "loses no messages under prefetch: stranded messages are all redelivered after the VT" $ \TestFixture {pool, queueName, dlqName = _} -> do
let total = 20 :: Int
vt = 4 :: Int32
forM_ [1 .. total] $ \i ->
runPgmqSession pool $ do
_ <-
Sessions.sendMessage $
SendMessage
{ queueName = queueName,
messageBody = MessageBody (String $ Text.pack ("noloss-" <> show i)),
delay = Just 0
}
pure ()
processedRef <- newIORef (0 :: Int)
let config =
(defaultConfig queueName)
{ visibilityTimeout = vt,
batchSize = 2,
polling = StandardPolling {pollInterval = 0.02},
prefetchConfig = Just defaultPrefetchConfig
}
-- Handler blocks for far longer than the test window, so messages are
-- read-ahead into the buffer/inbox but never acked before shutdown.
_ <- timeout 30_000_000 $ runAdapterIO pool $ runTracingNoop $ do
adapter <- requireAdapter pool config
let handler = slowHandler processedRef 10_000_000
processor = mkProcessor adapter handler
appResult <- runApp (AppConfig {strategy = IgnoreFailures, inboxSize = 2}) [(ProcessorId "noloss-test", processor)]
case appResult of
Left err -> liftIO $ expectationFailure $ "Failed to start app: " <> show err
Right appHandle -> do
liftIO $ threadDelay 1_000_000 -- let the adapter read a few batches ahead
_ <- stopAppGracefully (ShutdownConfig {drainTimeout = 1}) appHandle
pure ()
processed <- readIORef processedRef
-- Wait past the visibility timeout so every read-but-stranded message
-- becomes visible again.
threadDelay (fromIntegral vt * 1_000_000 + 2_000_000)
recoverable <-
runPgmqSession pool $ do
msgs <-
Sessions.readMessage $
ReadMessage
{ queueName = queueName,
delay = 30,
batchSize = Just 200,
conditional = Nothing
}
pure $ Vector.length msgs
liftIO $
putStrLn $
"PREFETCH-NOLOSS-DIAG: total="
<> show total
<> " processed="
<> show processed
<> " recoverableAfterVT="
<> show recoverable
-- No message is lost: everything not AckOk-deleted is back in the queue.
(processed + recoverable) `shouldBe` total
--------------------------------------------------------------------------------
-- Test Helpers
--------------------------------------------------------------------------------
-- | Handler that immediately dead-letters messages.
deadLetterHandler :: (IOE :> es) => IORef Int -> Handler es Value
deadLetterHandler processedRef _ = do
liftIO $ atomicModifyIORef' processedRef (\n -> (n + 1, ()))
pure $ AckDeadLetter (PoisonPill "Test dead-letter")
-- | Handler that takes a specified delay before acking.
slowHandler :: (IOE :> es) => IORef Int -> Int -> Handler es Value
slowHandler processedRef delayMicros _ = do
liftIO $ threadDelay delayMicros
liftIO $ atomicModifyIORef' processedRef (\n -> (n + 1, ()))
pure AckOk
-- | Handler that just counts messages.
countingHandler :: (IOE :> es) => IORef Int -> Handler es Value
countingHandler processedRef _ = do
liftIO $ atomicModifyIORef' processedRef (\n -> (n + 1, ()))
pure AckOk
-- | Wait until the processed count reaches the target, with timeout.
waitForProcessed :: IORef Int -> Int -> Int -> IO ()
waitForProcessed ref target timeoutMicros = go 0
where
pollInterval = 50000 -- 50ms
go elapsed
| elapsed >= timeoutMicros = pure ()
| otherwise = do
count <- readIORef ref
if count >= target
then pure ()
else do
threadDelay pollInterval
go (elapsed + pollInterval)
requireAdapter ::
(Pgmq :> es, Error PgmqRuntimeError :> es, IOE :> es, Tracing :> es) =>
Pool.Pool ->
PgmqAdapterConfig ->
Eff es (Adapter es Value)
requireAdapter pool config = do
result <- pgmqAdapter (mkPgmqAdapterEnv pool) config
case result of
Left err -> liftIO $ error $ "Invalid PGMQ adapter config: " <> show err
Right adapter -> pure adapter
-- | Send a batch of messages, run a slow-handler processor against @q@ until a
-- couple are processed, then shut down and measure how many messages remain in
-- the queue but invisible. Returns @(total, processed, visibleAfterShutdown)@.
--
-- Parameters are chosen so 'parBuffered' can buffer several small chunks ahead
-- of the shutdown gate: small @batchSize@, default @bufferSize@ (4), a slow
-- handler, and a small inbox — so the buffer is non-empty at shutdown.
measureShutdownRelease :: Pool.Pool -> QueueName -> Maybe PrefetchConfig -> IO (Int, Int, Int)
measureShutdownRelease pool q mprefetch = do
let total = 40 :: Int
forM_ [1 .. total] $ \i ->
runPgmqSession pool $ do
_ <-
Sessions.sendMessage $
SendMessage
{ queueName = q,
messageBody = MessageBody (String $ Text.pack ("shutdown-" <> show i)),
delay = Just 0
}
pure ()
processedRef <- newIORef (0 :: Int)
let config =
(defaultConfig q)
{ visibilityTimeout = 30,
batchSize = 2,
polling = StandardPolling {pollInterval = 0.02},
prefetchConfig = mprefetch
}
_ <- timeout 30_000_000 $ runAdapterIO pool $ runTracingNoop $ do
adapter <- requireAdapter pool config
let handler = slowHandler processedRef 400000 -- 0.4s per message
processor = mkProcessor adapter handler
appResult <- runApp (AppConfig {strategy = IgnoreFailures, inboxSize = 2}) [(ProcessorId "shutdown-measure", processor)]
case appResult of
Left err -> liftIO $ expectationFailure $ "Failed to start app: " <> show err
Right appHandle -> do
liftIO $ waitForProcessed processedRef 1 5_000_000
_ <- stopAppGracefully (ShutdownConfig {drainTimeout = 1}) appHandle
pure ()
processed <- readIORef processedRef
visibleNow <-
runPgmqSession pool $ do
msgs <-
Sessions.readMessage $
ReadMessage
{ queueName = q,
delay = 30,
batchSize = Just 100,
conditional = Nothing
}
pure $ Vector.length msgs
pure (total, processed, visibleNow)