diff --git a/CHANGELOG.md b/CHANGELOG.md --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -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 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.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, 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 @@ -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) 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 @@ -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. diff --git a/test/Shibuya/Adapter/Pgmq/ChaosSpec.hs b/test/Shibuya/Adapter/Pgmq/ChaosSpec.hs --- a/test/Shibuya/Adapter/Pgmq/ChaosSpec.hs +++ b/test/Shibuya/Adapter/Pgmq/ChaosSpec.hs @@ -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 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 @@ -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 diff --git a/test/TmpPostgres.hs b/test/TmpPostgres.hs --- a/test/TmpPostgres.hs +++ b/test/TmpPostgres.hs @@ -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). --