shibuya-pgmq-adapter 0.16.0.0 → 0.16.1.0
raw patch · 7 files changed
+435/−45 lines, 7 filesdep ~effectful-coredep ~pg-migratedep ~shibuya-corePVP ok
version bump matches the API change (PVP)
Dependency ranges changed: effectful-core, pg-migrate, shibuya-core
API changes (from Hackage documentation)
+ Shibuya.Adapter.Pgmq: PgmqAcknowledgementException :: PgmqRuntimeError -> PgmqAcknowledgementException
+ Shibuya.Adapter.Pgmq: newtype PgmqAcknowledgementException
Files
- CHANGELOG.md +34/−0
- shibuya-pgmq-adapter.cabal +6/−6
- src/Shibuya/Adapter/Pgmq.hs +2/−1
- src/Shibuya/Adapter/Pgmq/Internal.hs +41/−16
- test/Shibuya/Adapter/Pgmq/ChaosSpec.hs +287/−18
- test/Shibuya/Adapter/Pgmq/InternalSpec.hs +31/−1
- test/TmpPostgres.hs +34/−3
CHANGELOG.md view
@@ -1,5 +1,39 @@ # Changelog +## 0.16.1.0 — 2026-09-21++### Reliability++- Dead-letter movement now claims the source row before producing the DLQ copy+ in the same PostgreSQL transaction. A retry after an ambiguous successful+ commit observes that the source is already absent and does not emit a second+ DLQ row; a failed send rolls the claim back and leaves the source recoverable.+- Acknowledgement handles serialize concurrent callers and release ownership+ on cancellation. Completed duplicate calls are no-ops, while failed or+ cancelled attempts remain retryable.+- Exhausted acknowledgement errors call `PgmqAdapterEnv.onAckFailure` and throw+ `PgmqAcknowledgementException` synchronously, allowing Shibuya to retain the+ delivery as a terminal lifecycle failure.++### Tests++- Added real PostgreSQL coverage for discarded commit confirmation, concurrent+ finalization, rollback/recoverability, automatic-DLQ failure, core lifecycle+ visibility, pool reconnection after database restart, and lease-expiry+ redelivery, plus deterministic cancellation and renewal-outage regressions.++### Other Changes++- Require `shibuya-core ^>=0.10.0.0` across the adapter and benchmark for the+ coordinated lifecycle release candidate.+- Accept `pg-migrate` 1.2 in test components and make the live restart fixture+ stream identities externally before failing on duplicate, missing,+ unexpected, or malformed deliveries, avoiding ledger-induced heap growth.+- Accept `effectful-core` 2.7.1.1 and later in addition to the 2.6 family across+ the library, tests, benchmark, and example. Releases 2.7.0.0 through 2.7.1.0+ are excluded because upstream records a per-operation performance regression+ for dynamically dispatched effects, including this adapter's `Pgmq` effect.+ ## 0.16.0.0 — 2026-09-16 ### Breaking Changes
shibuya-pgmq-adapter.cabal view
@@ -1,6 +1,6 @@ cabal-version: 3.12 name: shibuya-pgmq-adapter-version: 0.16.0.0+version: 0.16.1.0 synopsis: PGMQ adapter for the Shibuya queue processing framework description: A Shibuya adapter that integrates with pgmq (PostgreSQL Message Queue)@@ -42,14 +42,14 @@ aeson ^>=2.2, base ^>=4.21.0.0, bytestring ^>=0.12,- effectful-core ^>=2.6.1.0,+ effectful-core (>=2.6.1 && <2.7) || (>=2.7.1.1 && <2.8), hasql ^>=1.10, hasql-pool ^>=1.4, hasql-transaction ^>=1.2, pgmq-core ^>=0.6, pgmq-effectful ^>=0.6, pgmq-hasql ^>=0.6,- shibuya-core ^>=0.9.0.0,+ shibuya-core ^>=0.10.0.0, stm ^>=2.5, streamly ^>=0.11, streamly-core ^>=0.3,@@ -106,20 +106,20 @@ async ^>=2.2, base ^>=4.21.0.0, bytestring,- effectful-core,+ effectful-core (>=2.6.1 && <2.7) || (>=2.7.1.1 && <2.8), ephemeral-pg, hasql ^>=1.10, hasql-pool ^>=1.4, hasql-transaction ^>=1.2, hspec ^>=2.11,- pg-migrate ^>=1.1,+ pg-migrate >=1.1 && <1.3, pgmq-core ^>=0.6, pgmq-effectful ^>=0.6, pgmq-hasql ^>=0.6, pgmq-migration ^>=0.6, quickcheck-instances ^>=0.3, random,- shibuya-core ^>=0.9.0.0,+ shibuya-core ^>=0.10.0.0, shibuya-pgmq-adapter, stm, streamly ^>=0.11,
src/Shibuya/Adapter/Pgmq.hs view
@@ -72,6 +72,7 @@ module Shibuya.Adapter.Pgmq ( -- * Adapter pgmqAdapter,+ PgmqAcknowledgementException (..), -- * Configuration PgmqAdapterConfig (..),@@ -166,7 +167,7 @@ topicDeadLetter, validateConfig, )-import Shibuya.Adapter.Pgmq.Internal (mkIngested, pgmqChunks, pgmqChunksPrefetch, releaseMessages)+import Shibuya.Adapter.Pgmq.Internal (PgmqAcknowledgementException (..), mkIngested, pgmqChunks, pgmqChunksPrefetch, releaseMessages) import Shibuya.Core.Ingested (Ingested) import Shibuya.Telemetry.Effect (Tracing) import Streamly.Data.Stream (Stream)
src/Shibuya/Adapter/Pgmq/Internal.hs view
@@ -13,6 +13,7 @@ finalizeAutoDeadLetter, -- * AckHandle Construction+ PgmqAcknowledgementException (..), mkAckHandle, mergeDlqHeaders, @@ -32,14 +33,16 @@ where import Control.Concurrent (threadDelay)-import Control.Monad (void, when)+import Control.Concurrent.MVar (newMVar, putMVar, takeMVar)+import Control.Exception qualified as BaseException+import Control.Monad (unless, void, when) import Control.Monad.IO.Class (liftIO) import Data.Aeson (Value (..)) import Data.Aeson.Key qualified as Key import Data.Aeson.KeyMap qualified as KeyMap import Data.Foldable (traverse_) import Data.Function ((&))-import Data.IORef (IORef, newIORef, readIORef, writeIORef)+import Data.IORef (atomicWriteIORef, newIORef, readIORef, writeIORef) import Data.Int (Int32) import Data.Text qualified as Text import Data.Text.Encoding qualified as TE@@ -56,6 +59,7 @@ (:>), ) import Effectful.Error.Static (Error, catchError, throwError)+import Effectful.Exception qualified as Exception import Hasql.Pool qualified as Pool import Hasql.Transaction qualified as Transaction import Hasql.Transaction.Sessions qualified as Transaction.Sessions@@ -118,6 +122,15 @@ import Streamly.Data.Stream.Prelude qualified as StreamP import Streamly.Data.Unfold qualified as Unfold +-- | A PGMQ acknowledgement operation failed after the adapter's bounded retry+-- budget. This synchronous exception crosses Shibuya's finalizer boundary so+-- core can retain the delivery as a lifecycle failure even after ingestion has+-- already stopped.+newtype PgmqAcknowledgementException = PgmqAcknowledgementException PgmqRuntimeError+ deriving stock (Show)++instance BaseException.Exception PgmqAcknowledgementException+ -- | Convert 'NominalDiffTime' to seconds as 'Int32', saturating at the -- 'Int32' bounds. --@@ -225,16 +238,23 @@ (Pgmq :> es, Error PgmqRuntimeError :> es, IOE :> es, Tracing :> es) => PgmqAdapterEnv -> PgmqAdapterConfig ->- IORef Bool -> Pgmq.Message ->- AckHandle es-mkAckHandle env config finalizedRef msg = AckHandle $ \decision -> do- alreadyFinalized <- liftIO $ readIORef finalizedRef- if alreadyFinalized- then pure ()- else do- runDecision decision- liftIO $ writeIORef finalizedRef True+ Eff es (AckHandle es)+mkAckHandle env config msg = do+ finalizerLock <- liftIO $ newMVar ()+ finalizedRef <- liftIO $ newIORef False+ pure $ AckHandle $ \decision ->+ Exception.bracket_+ (liftIO $ takeMVar finalizerLock)+ (liftIO $ putMVar finalizerLock ())+ $ do+ alreadyFinalized <- liftIO $ readIORef finalizedRef+ unless alreadyFinalized $ do+ runDecision decision+ `catchError` \_callStack err -> do+ liftIO $ env.onAckFailure msg err+ Exception.throwIO (PgmqAcknowledgementException err)+ liftIO $ atomicWriteIORef finalizedRef True where queueName = config.queueName msgId = msg.messageId@@ -297,7 +317,12 @@ Transaction.Sessions.Write tx tx = do- case dlqConfig.dlqTarget of+ -- Claim the source row before producing the DLQ copy. If the transaction+ -- committed but its response was lost, a retry observes False and emits+ -- nothing. If the send fails, PostgreSQL rolls the delete back with the+ -- rest of the transaction, leaving the source recoverable.+ claimed <- Transaction.statement sourceQuery Msg.deleteMessage+ when claimed $ case dlqConfig.dlqTarget of DirectQueue dlqQueueName -> case dlqHeaders of Just headers ->@@ -340,7 +365,6 @@ delay = Nothing } Msg.sendTopic- void $ Transaction.statement sourceQuery Msg.deleteMessage -- | Merge the consumer's current trace headers with the original -- message's headers JSON for the DLQ-write path.@@ -403,9 +427,8 @@ Pgmq.Message -> Eff es (Maybe (Ingested es Value)) mkIngested env config msg = do- finalizedRef <- liftIO $ newIORef False lease <- mkLease config msg- let ackHandle = mkAckHandle env config finalizedRef msg+ ackHandle <- mkAckHandle env config msg -- Check if max retries exceeded if msg.readCount > config.maxRetries then do@@ -435,7 +458,9 @@ Eff es () finalizeAutoDeadLetter msg onAutoDeadLetter onAckFailure finalizeAction = (finalizeAction >> liftIO (onAutoDeadLetter msg))- `catchError` \_callStack err -> liftIO (onAckFailure msg err)+ `catchError` \_callStack err -> do+ liftIO $ onAckFailure msg err+ throwError err -- | Stream of message batches from pgmq. -- Each element is a Vector of messages from a single poll.
test/Shibuya/Adapter/Pgmq/ChaosSpec.hs view
@@ -10,15 +10,18 @@ module Shibuya.Adapter.Pgmq.ChaosSpec (spec) where import Control.Concurrent (threadDelay)-import Control.Concurrent.Async (async, cancel)-import Control.Monad (forM_)+import Control.Concurrent.Async (async, cancel, concurrently_)+import Control.Concurrent.MVar (newEmptyMVar, putMVar, takeMVar)+import Control.Exception qualified as Exception+import Control.Monad (forM_, when) 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 (Eff, IOE, Limit (..), Persistence (..), UnliftStrategy (..), liftIO, runEff, withEffToIO, (:>))+import Effectful.Dispatch.Dynamic (interpret) import Effectful.Error.Static (Error, runErrorNoCallStack) import Hasql.Decoders qualified as D import Hasql.Pool qualified as Pool@@ -26,13 +29,17 @@ import Hasql.Statement qualified as Statement import Pgmq.Effectful (Pgmq, PgmqRuntimeError, runPgmq) import Pgmq.Effectful qualified as PgmqEff+import Pgmq.Effectful.Effect qualified as PgmqEffect import Pgmq.Hasql.Encoders qualified as Encoders 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 Pgmq.Types (Message (..), MessageBody (..), MessageHeaders (..), QueueName, parseQueueName)+import Shibuya.Adapter (Adapter (..)) import Shibuya.Adapter.Pgmq- ( PgmqAdapterConfig (..),+ ( PgmqAcknowledgementException,+ PgmqAdapterConfig (..),+ PgmqAdapterEnv (..),+ PollRetryConfig (..), PollingConfig (..), PrefetchConfig (..), defaultConfig,@@ -41,26 +48,31 @@ mkPgmqAdapterEnv, pgmqAdapter, )-import Shibuya.Adapter.Pgmq.Internal (mkIngested)+import Shibuya.Adapter.Pgmq.Internal (mkAckHandle, mkIngested) import Shibuya.App ( AppConfig (..), ProcessorId (..), ShutdownConfig (..), SupervisionStrategy (..), defaultAppConfig,+ defaultShutdownConfig,+ getAppMaster, mkProcessor, runApp, stopAppGracefully,+ waitApp, ) import Shibuya.Core.Ack (AckDecision (..), DeadLetterCode, DeadLetterReason (..), HaltReason (..), mkDeadLetterCode) import Shibuya.Core.AckHandle (AckHandle (..)) import Shibuya.Core.Ingested (Ingested (..)) import Shibuya.Handler (Handler)+import Shibuya.Internal.Runner.Master (getLifecycleSnapshot) import Shibuya.Telemetry.Effect (Tracing, runTracingNoop)+import Streamly.Data.Stream qualified as Stream import System.Environment (lookupEnv) import System.Timeout (timeout) import Test.Hspec-import TmpPostgres (TestFixture (..), runPgmqSession, withPgmqDb, withTestFixture)+import TmpPostgres (TestFixture (..), runPgmqSession, withPgmqDb, withRestartablePgmqDb, withTestFixture) spec :: Spec spec = do@@ -70,7 +82,31 @@ longHandlerSpec gracefulShutdownSpec prefetchSpec+ describe "Database restart" $ do+ it "reconnects and redelivers an unacknowledged message after its lease expires" $ do+ result <- withRestartablePgmqDb $ \pool restartDatabase ->+ withTestFixture pool $ \TestFixture {queueName} -> do+ runPgmqSession pool $ do+ _ <- Sessions.sendMessage $ SendMessage queueName (MessageBody (String "restart-redelivery")) (Just 0)+ pure ()+ leased <- runPgmqSession pool $ Sessions.readMessage $ ReadMessage queueName 1 (Just 1) Nothing+ originalId <- case Vector.uncons leased of+ Nothing -> expectationFailure "expected one leased message before restart" >> pure Nothing+ Just (msg, _) -> pure $ Just msg.messageId + restartDatabase >>= \case+ Left err -> expectationFailure $ "ephemeral PostgreSQL restart failed: " <> show err+ Right () -> pure ()++ threadDelay 1_500_000+ redelivered <- runPgmqSession pool $ Sessions.readMessage $ ReadMessage queueName 30 (Just 1) Nothing+ case (originalId, Vector.uncons redelivered) of+ (Just expectedId, Just (msg, _)) -> msg.messageId `shouldBe` expectedId+ _ -> expectationFailure "expected the same durable message after restart and lease expiry"+ case result of+ Left err -> expectationFailure $ "failed to start ephemeral PostgreSQL: " <> show err+ Right () -> pure ()+ -- | Wrapper to run tests with a temporary database and fixture withTempDbFixture :: (TestFixture -> IO ()) -> IO () withTempDbFixture action = do@@ -181,7 +217,7 @@ liftIO $ waitForProcessed processedRef 1 3000000 -- Stop the app- let shutdownConfig = ShutdownConfig {drainTimeout = 5}+ let shutdownConfig = defaultShutdownConfig {drainTimeout = 5} _ <- stopAppGracefully shutdownConfig appHandle pure () @@ -242,7 +278,7 @@ Left err -> liftIO $ expectationFailure $ "Failed to start app: " <> show err Right appHandle -> do liftIO $ waitForProcessed processedRef 1 3000000- _ <- stopAppGracefully ShutdownConfig {drainTimeout = 5} appHandle+ _ <- stopAppGracefully defaultShutdownConfig {drainTimeout = 5} appHandle pure () applicationInspection <- runPgmqSession pool $ inspectNextDlqPayload dlqName@@ -340,7 +376,7 @@ liftIO $ waitForProcessed processedRef 1 3000000 -- Stop the app- let shutdownConfig = ShutdownConfig {drainTimeout = 5}+ let shutdownConfig = defaultShutdownConfig {drainTimeout = 5} _ <- stopAppGracefully shutdownConfig appHandle pure () @@ -430,6 +466,212 @@ } Vector.length sourceMsgs `shouldBe` 0 + it "a discarded DLQ commit confirmation can be retried without creating a second copy" $ \TestFixture {pool, queueName, dlqName} -> do+ runPgmqSession pool $ do+ _ <-+ Sessions.sendMessage $+ SendMessage+ { queueName = queueName,+ messageBody = MessageBody (String "ambiguous-dlq-commit"),+ delay = Just 0+ }+ pure ()++ let config =+ (defaultConfig queueName)+ { visibilityTimeout = 5,+ batchSize = 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+ let finalizeWithFreshHandle = do+ ingestedResult <- mkIngested (mkPgmqAdapterEnv pool) config msg+ case ingestedResult of+ Nothing -> liftIO $ expectationFailure "message should not auto-DLQ"+ Just Ingested {ack = AckHandle finalize} ->+ finalize (AckDeadLetter (PoisonPill "ambiguous commit"))++ -- Treat the first successful return as a commit confirmation that the+ -- caller never received: discard its in-memory handle and retry the+ -- same durable delivery through a fresh handle.+ finalizeWithFreshHandle+ finalizeWithFreshHandle++ dlqMsgs <-+ runPgmqSession pool $+ Sessions.readMessage $+ ReadMessage+ { queueName = dlqName,+ delay = 30,+ batchSize = Just 10,+ conditional = Nothing+ }+ Vector.length dlqMsgs `shouldBe` 1++ sourceMetrics <- runPgmqSession pool $ Sessions.queueMetrics queueName+ sourceMetrics.queueLength `shouldBe` 0++ it "concurrent finalization of one delivery converges on one durable DLQ move" $ \TestFixture {pool, queueName, dlqName} -> do+ runPgmqSession pool $ do+ _ <- Sessions.sendMessage $ SendMessage queueName (MessageBody (String "concurrent-dlq")) (Just 0)+ pure ()++ let config =+ (defaultConfig queueName)+ { deadLetterConfig = Just $ directDeadLetter dlqName True+ }++ runAdapterIO pool $ runTracingNoop $ do+ msgs <- PgmqEff.readMessage $ ReadMessage queueName 30 (Just 1) 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} ->+ withEffToIO (ConcUnlift Persistent Unlimited) $ \runInIO ->+ concurrently_+ (runInIO $ finalize $ AckDeadLetter $ PoisonPill "first")+ (runInIO $ finalize $ AckDeadLetter $ PoisonPill "second")++ dlqMetrics <- runPgmqSession pool $ Sessions.queueMetrics dlqName+ dlqMetrics.queueLength `shouldBe` 1+ sourceMetrics <- runPgmqSession pool $ Sessions.queueMetrics queueName+ sourceMetrics.queueLength `shouldBe` 0++ it "cancellation releases finalizer ownership so the same handle can retry" $ \TestFixture {pool, queueName, dlqName = _} -> do+ runPgmqSession pool $ do+ _ <- Sessions.sendMessage $ SendMessage queueName (MessageBody (String "cancelled-finalize")) (Just 0)+ pure ()++ msgs <- runPgmqSession pool $ Sessions.readMessage $ ReadMessage queueName 30 (Just 1) Nothing+ case Vector.uncons msgs of+ Nothing -> expectationFailure "expected one source message"+ Just (msg, _) -> do+ started <- newEmptyMVar+ release <- newEmptyMVar+ blockAttempt <- newIORef True+ attempts <- newIORef (0 :: Int)+ let config = defaultConfig queueName+ runWithBlockingDelete ::+ Eff '[Tracing, Pgmq, Error PgmqRuntimeError, IOE] a ->+ IO (Either PgmqRuntimeError a)+ runWithBlockingDelete action =+ runEff+ $ runErrorNoCallStack+ $ interpret+ ( \_ -> \case+ PgmqEffect.DeleteMessage _ -> do+ shouldBlock <- liftIO $ atomicModifyIORef' blockAttempt (\b -> (False, b))+ liftIO $ atomicModifyIORef' attempts (\n -> (n + 1, ()))+ when shouldBlock $ liftIO $ putMVar started () >> takeMVar release+ pure True+ _ -> error "unexpected PGMQ operation in cancellation regression"+ )+ $ runTracingNoop action++ result <- runWithBlockingDelete $ do+ AckHandle finalize <- mkAckHandle (mkPgmqAdapterEnv pool) config msg+ withEffToIO (ConcUnlift Persistent Unlimited) $ \runInIO -> do+ worker <- async $ runInIO $ finalize AckOk+ takeMVar started+ cancel worker+ timeout 1_000_000 $ runInIO $ finalize AckOk++ result `shouldBe` Right (Just ())+ readIORef attempts `shouldReturn` 2++ it "a failed DLQ move remains recoverable and reaches the core lifecycle" $ \TestFixture {pool, queueName, dlqName = _} -> do+ runPgmqSession pool $ do+ _ <- Sessions.sendMessage $ SendMessage queueName (MessageBody (String "failed-dlq-move")) (Just 0)+ pure ()++ failureCalls <- newIORef (0 :: Int)+ let missingDlq = queueNameOrFail "missing_dlq_target"+ config =+ (defaultConfig queueName)+ { deadLetterConfig = Just $ directDeadLetter missingDlq True,+ ackRetry = PollRetryConfig 1 0 0+ }+ env =+ (mkPgmqAdapterEnv pool)+ { onAckFailure = \_ _ -> bump failureCalls+ }+ processorId = ProcessorId "pgmq-terminal-ack-failure"++ lifecycle <- runAdapterIO pool $ runTracingNoop $ do+ msgs <- PgmqEff.readMessage $ ReadMessage queueName 30 (Just 1) Nothing+ case Vector.uncons msgs of+ Nothing -> liftIO (expectationFailure "expected one source message") >> pure ""+ Just (msg, _) -> do+ ingestedResult <- mkIngested env config msg+ case ingestedResult of+ Nothing -> liftIO (expectationFailure "message should not auto-DLQ") >> pure ""+ Just ingested -> do+ let adapter = Adapter "pgmq:test-terminal-ack-failure" (Stream.fromList [ingested]) (pure ())+ processor = mkProcessor adapter (\_ -> pure $ AckDeadLetter $ PoisonPill "fail loudly")+ appResult <- runApp defaultAppConfig [(processorId, processor)]+ case appResult of+ Left appError -> liftIO $ error $ "runApp failed: " <> show appError+ Right appHandle -> do+ waitApp appHandle+ snapshot <- getLifecycleSnapshot (getAppMaster appHandle)+ pure $ show snapshot++ lifecycle `shouldContain` "LifecycleFailed"+ lifecycle `shouldContain` "pgmq-terminal-ack-failure"+ readIORef failureCalls `shouldReturn` 4+ sourceMetrics <- runPgmqSession pool $ Sessions.queueMetrics queueName+ sourceMetrics.queueLength `shouldBe` 1++ it "a failed automatic DLQ move calls the failure hook and remains visible" $ \TestFixture {pool, queueName, dlqName = _} -> do+ runPgmqSession pool $ do+ _ <- Sessions.sendMessage $ SendMessage queueName (MessageBody (String "failed-auto-dlq")) (Just 0)+ pure ()++ msgs <- runPgmqSession pool $ Sessions.readMessage $ ReadMessage queueName 30 (Just 1) Nothing+ case Vector.uncons msgs of+ Nothing -> expectationFailure "expected one source message"+ Just (msg, _) -> do+ failureCalls <- newIORef (0 :: Int)+ autoCalls <- newIORef (0 :: Int)+ let config =+ (defaultConfig queueName)+ { maxRetries = 0,+ deadLetterConfig = Just $ directDeadLetter (queueNameOrFail "missing_auto_dlq_target") True,+ ackRetry = PollRetryConfig 1 0 0+ }+ env =+ (mkPgmqAdapterEnv pool)+ { onAutoDeadLetter = \_ -> bump autoCalls,+ onAckFailure = \_ _ -> bump failureCalls+ }+ attempt = runAdapterIO pool $ runTracingNoop $ do+ _ <- mkIngested env config msg+ pure ()++ outcome <- (Exception.try attempt :: IO (Either PgmqAcknowledgementException ()))+ case outcome of+ Left _ -> pure ()+ Right () -> expectationFailure "expected automatic DLQ failure to be visible"+ readIORef autoCalls `shouldReturn` 0+ readIORef failureCalls `shouldReturn` 1+ sourceMetrics <- runPgmqSession pool $ Sessions.queueMetrics queueName+ sourceMetrics.queueLength `shouldBe` 1+ it "AckOk is idempotent after a successful finalize" $ \TestFixture {pool, queueName, dlqName = _} -> do runPgmqSession pool $ do _ <-@@ -562,7 +804,7 @@ liftIO $ waitForProcessed processedRef 1 5000000 -- Stop the app- let shutdownConfig = ShutdownConfig {drainTimeout = 5}+ let shutdownConfig = defaultShutdownConfig {drainTimeout = 5} _ <- stopAppGracefully shutdownConfig appHandle pure () @@ -610,10 +852,29 @@ Left err -> liftIO $ expectationFailure ("Failed to start app: " <> show err) >> pure False Right appHandle -> do liftIO $ threadDelay 100000- stopAppGracefully ShutdownConfig {drainTimeout = 2} appHandle+ stopAppGracefully defaultShutdownConfig {drainTimeout = 2} appHandle drained `shouldBe` True + it "returns the same completed result for a repeated shutdown" $ \TestFixture {pool, queueName, dlqName = _} -> do+ let config =+ (defaultConfig queueName)+ { polling = StandardPolling {pollInterval = 0.1}+ }++ outcomes <- runAdapterIO pool $ runTracingNoop $ do+ adapter <- requireAdapter pool config+ appResult <- runApp defaultAppConfig [(ProcessorId "repeated-stop-test", mkProcessor adapter (\_ -> pure AckOk))]+ case appResult of+ Left err -> liftIO $ expectationFailure ("Failed to start app: " <> show err) >> pure (False, False)+ Right appHandle -> do+ let shutdownConfig = defaultShutdownConfig {drainTimeout = 2}+ first <- stopAppGracefully shutdownConfig appHandle+ second <- stopAppGracefully shutdownConfig appHandle+ pure (first, second)++ outcomes `shouldBe` (True, True)+ it "processes in-flight messages during shutdown" $ \TestFixture {pool, queueName, dlqName = _} -> do -- Send multiple messages forM_ [1 .. 5 :: Int] $ \i ->@@ -650,7 +911,7 @@ liftIO $ waitForProcessed processedRef 5 5000000 -- Graceful shutdown- let shutdownConfig = ShutdownConfig {drainTimeout = 2}+ let shutdownConfig = defaultShutdownConfig {drainTimeout = 2} _ <- stopAppGracefully shutdownConfig appHandle pure () @@ -705,7 +966,7 @@ pure () -- Shutdown quickly- let shutdownConfig = ShutdownConfig {drainTimeout = 1}+ let shutdownConfig = defaultShutdownConfig {drainTimeout = 1} _ <- stopAppGracefully shutdownConfig appHandle pure () @@ -778,7 +1039,7 @@ 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}+ let shutdownConfig = defaultShutdownConfig {drainTimeout = 5} _ <- stopAppGracefully shutdownConfig appHandle pure () @@ -906,7 +1167,7 @@ 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+ _ <- stopAppGracefully (defaultShutdownConfig {drainTimeout = 1}) appHandle pure () processed <- readIORef processedRef@@ -998,6 +1259,14 @@ liftIO $ atomicModifyIORef' processedRef (\n -> (n + 1, ())) pure AckOk +bump :: IORef Int -> IO ()+bump ref = atomicModifyIORef' ref (\n -> (n + 1, ()))++queueNameOrFail :: Text.Text -> QueueName+queueNameOrFail raw = case parseQueueName raw of+ Left err -> error $ "invalid test queue name: " <> show err+ Right name -> name+ -- | Wait until the processed count reaches the target, with timeout. waitForProcessed :: IORef Int -> Int -> Int -> IO () waitForProcessed ref target timeoutMicros = go 0@@ -1064,7 +1333,7 @@ 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+ _ <- stopAppGracefully (defaultShutdownConfig {drainTimeout = 1}) appHandle pure () processed <- readIORef processedRef
test/Shibuya/Adapter/Pgmq/InternalSpec.hs view
@@ -30,12 +30,14 @@ import Shibuya.Adapter.Pgmq.Internal ( finalizeAutoDeadLetter, mergeDlqHeaders,+ mkLease, mkReadGrouped, mkReadMessage, mkReadWithPoll, nominalToSeconds, pgmqChunks, )+import Shibuya.Core.Lease (Lease (..)) import Streamly.Data.Stream qualified as Stream import Test.Hspec @@ -47,6 +49,7 @@ mkReadGroupedSpec fifoDispatchSpec pollRetrySpec+ leaseRetrySpec autoDeadLetterHookSpec mergeDlqHeadersSpec @@ -330,6 +333,33 @@ result `shouldBe` Left transientError readIORef calls `shouldReturn` 2 +leaseRetrySpec :: Spec+leaseRetrySpec = describe "lease renewal retry" $ do+ it "recovers from a transient renewal outage within the configured budget" $ do+ calls <- newIORef (0 :: Int)+ let cfg = (retryTestConfig 1) {ackRetry = PollRetryConfig 3 0 0}+ action :: Eff '[Pgmq, Error PgmqRuntimeError, IOE] ()+ action = do+ lease <- mkLease cfg testMessage+ lease.leaseExtend 30++ result <-+ runEff $+ runErrorNoCallStack $+ interpret+ ( \_ -> \case+ PgmqEffect.SetVisibilityTimeoutAt _ -> do+ attempt <- liftIO $ atomicModifyIORef' calls (\n -> let next = n + 1 in (next, next))+ if attempt < 3+ then throwError transientError+ else pure $ Just testMessage+ _ -> error "unexpected PGMQ operation in lease renewal retry test"+ )+ action++ result `shouldBe` Right ()+ readIORef calls `shouldReturn` 3+ autoDeadLetterHookSpec :: Spec autoDeadLetterHookSpec = describe "finalizeAutoDeadLetter" $ do it "calls the auto-DLQ hook only after finalize succeeds" $ do@@ -362,7 +392,7 @@ result <- runEff $ runErrorNoCallStack action - result `shouldBe` Right ()+ result `shouldBe` Left permanentError readIORef autoCalls `shouldReturn` 0 readIORef failureCalls `shouldReturn` 1
test/TmpPostgres.hs view
@@ -5,6 +5,7 @@ module TmpPostgres ( -- * Test Execution withPgmqDb,+ withRestartablePgmqDb, withTestFixture, -- * Test Fixture@@ -16,13 +17,16 @@ where import Control.Exception (bracket)+import Control.Monad ((>=>))+import Data.IORef (newIORef, readIORef, writeIORef) import Data.List.NonEmpty (NonEmpty ((:|))) import Data.Text (Text) import Data.Text qualified as Text import Data.Time (secondsToDiffTime) import Data.Word (Word64) import Database.PostgreSQL.Migrate qualified as Migrate-import EphemeralPg (StartError, connectionSettings, with)+import EphemeralPg (StartError)+import EphemeralPg qualified as Pg import Hasql.Connection.Settings qualified as Settings import Hasql.Pool qualified as Pool import Hasql.Pool.Config qualified as PoolConfig@@ -45,8 +49,8 @@ -- This creates an ephemeral PostgreSQL instance, installs the pgmq schema, -- and then runs the provided action with a connection pool. withPgmqDb :: (Pool.Pool -> IO a) -> IO (Either StartError a)-withPgmqDb action = with $ \db -> do- let connSettings = connectionSettings db+withPgmqDb action = Pg.with $ \db -> do+ let connSettings = Pg.connectionSettings db -- Install pgmq schema installPgmqSchema connSettings@@ -56,6 +60,33 @@ (createPool connSettings) Pool.release action++-- | Run against an ephemeral database that the test may restart in place.+-- The restart preserves the data directory, releases stale pooled connections,+-- and lets the next pool use establish a fresh connection to the same socket.+withRestartablePgmqDb ::+ (Pool.Pool -> IO (Either StartError ()) -> IO a) ->+ IO (Either StartError a)+withRestartablePgmqDb action = Pg.with $ \initialDb ->+ bracket+ (newIORef initialDb)+ (readIORef >=> Pg.stop)+ $ \databaseRef -> do+ let connSettings = Pg.connectionSettings initialDb+ installPgmqSchema connSettings+ bracket+ (createPool connSettings)+ Pool.release+ $ \pool -> do+ let restartDatabase = do+ currentDb <- readIORef databaseRef+ Pg.restart currentDb >>= \case+ Left err -> pure $ Left err+ Right restartedDb -> do+ writeIORef databaseRef restartedDb+ Pool.release pool+ pure $ Right ()+ action pool restartDatabase -- | Run an action with a test fixture (pool + unique queue names). --