packages feed

shibuya-pgmq-adapter 0.11.0.0 → 0.12.0.0

raw patch · 5 files changed

+74/−20 lines, 5 filesdep +pg-migratedep ~pgmq-coredep ~pgmq-effectfuldep ~pgmq-hasqlPVP ok

version bump matches the API change (PVP)

Dependencies added: pg-migrate

Dependency ranges changed: pgmq-core, pgmq-effectful, pgmq-hasql, pgmq-migration

API changes (from Hackage documentation)

Files

CHANGELOG.md view
@@ -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`.
shibuya-pgmq-adapter.cabal view
@@ -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,
src/Shibuya/Adapter/Pgmq.hs view
@@ -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
test/Shibuya/Adapter/Pgmq/ChaosSpec.hs view
@@ -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 ->
test/TmpPostgres.hs view
@@ -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