diff --git a/CHANGELOG.md b/CHANGELOG.md --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,35 @@ # Changelog +## 0.12.0.0 — 2026-07-14 + +Driven by the `pgmq-hs` 0.4 release. Still paired with `shibuya-core 0.8.0.1` +(unchanged bound). + +### Breaking Changes + +- Requires the `pgmq-*` 0.4 package family: `pgmq-core ^>=0.4`, `pgmq-effectful ^>=0.4`, + and `pgmq-hasql ^>=0.4` in the library (plus `pgmq-migration ^>=0.4` and + `pg-migrate ^>=1.1` in the test stanza), up from `^>=0.3`. +- The adapter's own API is unchanged, but consumers that install the PGMQ schema with + `pgmq-migration` must migrate: 0.4 replaced the `migrate` / `upgrade` / `validate` + runner with a `pg-migrate` component. `Pgmq.Migration` now exports only + `pgmqMigrations`, `MigrationComponent`, and `DefinitionError`; an application composes + the component into a plan with `migrationPlan` and runs it with `runMigrationPlan`, + which takes hasql connection settings rather than a `Pool` or `Connection`. +- A database created by `pgmq-migration` 0.3 or earlier must have its + `public.schema_migrations` ledger imported once via + `Pgmq.Migration.History.HasqlMigration` before the native runner takes over — the + native runner keeps its own ledger and does not read the old table. See + [Installing the PGMQ schema](../docs/user/pgmq-getting-started.md#installing-the-pgmq-schema). + +### Bug Fixes + +- Idle streams now observe shutdown. The shutdown gate was checked only *after* + empty poll chunks were filtered out, so a processor with nothing to consume kept + polling until it was forcibly cancelled instead of finishing on request. The gate + is now checked before the empty-chunk filter, covered by a PostgreSQL lifecycle + regression test. + ## 0.11.0.0 — 2026-07-04 Paired with `shibuya-core 0.8.0.1`. 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.11.0.0 +version: 0.12.0.0 synopsis: PGMQ adapter for the Shibuya queue processing framework description: A Shibuya adapter that integrates with pgmq (PostgreSQL Message Queue) @@ -46,9 +46,9 @@ hasql ^>=1.10, hasql-pool ^>=1.4, hasql-transaction ^>=1.2, - pgmq-core ^>=0.3, - pgmq-effectful ^>=0.3, - pgmq-hasql ^>=0.3, + pgmq-core ^>=0.4, + pgmq-effectful ^>=0.4, + pgmq-hasql ^>=0.4, shibuya-core ^>=0.8.0.1, stm ^>=2.5, streamly ^>=0.11, @@ -112,10 +112,11 @@ hasql-pool ^>=1.4, hasql-transaction ^>=1.2, hspec ^>=2.11, - pgmq-core ^>=0.3, - pgmq-effectful ^>=0.3, - pgmq-hasql ^>=0.3, - pgmq-migration ^>=0.3, + pg-migrate ^>=1.1, + pgmq-core ^>=0.4, + pgmq-effectful ^>=0.4, + pgmq-hasql ^>=0.4, + pgmq-migration ^>=0.4, quickcheck-instances ^>=0.3, random, shibuya-core ^>=0.8.0.1, 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 @@ -227,8 +227,8 @@ Stream (Eff es) (Ingested es Value) pgmqSourceWithShutdown env config shutdownVar = chunkStream - & Stream.filter (not . Vector.null) & Stream.takeWhileM keepChunk + & Stream.filter (not . Vector.null) & Stream.unfoldEach (Unfold.unfoldr Vector.uncons) & Stream.mapMaybeM (mkIngested env config) where 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 @@ -502,6 +502,25 @@ -- | Tests for graceful shutdown behavior. gracefulShutdownSpec :: SpecWith TestFixture gracefulShutdownSpec = describe "Graceful shutdown" $ do + it "drains an idle queue instead of waiting for the timeout" $ \TestFixture {pool, queueName, dlqName = _} -> do + let config = + (defaultConfig queueName) + { visibilityTimeout = 30, + batchSize = 1, + polling = StandardPolling {pollInterval = 0.1} + } + + drained <- runAdapterIO pool $ runTracingNoop $ do + adapter <- requireAdapter pool config + appResult <- runApp defaultAppConfig [(ProcessorId "idle-drain-test", mkProcessor adapter (\_ -> pure AckOk))] + case appResult of + Left err -> liftIO $ expectationFailure ("Failed to start app: " <> show err) >> pure False + Right appHandle -> do + liftIO $ threadDelay 100000 + stopAppGracefully ShutdownConfig {drainTimeout = 2} appHandle + + drained `shouldBe` True + it "processes in-flight messages during shutdown" $ \TestFixture {pool, queueName, dlqName = _} -> do -- Send multiple messages forM_ [1 .. 5 :: Int] $ \i -> diff --git a/test/TmpPostgres.hs b/test/TmpPostgres.hs --- a/test/TmpPostgres.hs +++ b/test/TmpPostgres.hs @@ -16,12 +16,13 @@ where import Control.Exception (bracket) +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 Hasql.Connection qualified as Connection import Hasql.Connection.Settings qualified as Settings import Hasql.Pool qualified as Pool import Hasql.Pool.Config qualified as PoolConfig @@ -101,18 +102,21 @@ Right a -> pure a -- | Install pgmq schema into a PostgreSQL database. +-- +-- The database is always fresh, so the native pg-migrate runner applies the +-- pgmq component from scratch; no predecessor ledger import is involved. installPgmqSchema :: Settings.Settings -> IO () installPgmqSchema connSettings = do - connResult <- Connection.acquire connSettings - case connResult of - Left err -> error $ "Failed to connect for migration: " <> show err - Right conn -> do - result <- Connection.use conn Migration.migrate - Connection.release conn - case result of - Left sessionErr -> error $ "Migration session error: " <> show sessionErr - Right (Left migrationErr) -> error $ "Migration error: " <> show migrationErr - Right (Right ()) -> pure () + component <- case Migration.pgmqMigrations of + Left defErr -> error $ "pgmq migration definition error: " <> show defErr + Right component -> pure component + plan <- case Migrate.migrationPlan (component :| []) of + Left planErr -> error $ "pgmq migration plan error: " <> show planErr + Right plan -> pure plan + result <- Migrate.runMigrationPlan Migrate.defaultRunOptions connSettings plan + case result of + Left migrationErr -> error $ "Migration error: " <> show migrationErr + Right _report -> pure () -- | Create a connection pool from connection settings. createPool :: Settings.Settings -> IO Pool.Pool