pgmq-migration 0.6.1.1 → 0.7.0.0
raw patch · 7 files changed
+314/−15 lines, 7 filesdep +postgresql-libpqdep ~ephemeral-pgdep ~pgmq-migrationPVP ok
version bump matches the API change (PVP)
Dependencies added: postgresql-libpq
Dependency ranges changed: ephemeral-pg, pgmq-migration
API changes (from Hackage documentation)
Files
- CHANGELOG.md +15/−0
- migrations/0007-notify-only-registered-queues.sql +74/−0
- migrations/manifest +1/−0
- pgmq-migration.cabal +8/−3
- test/Main.hs +105/−12
- test/NotifyListener.hs +48/−0
- test/RootPermissionsSpec.hs +63/−0
CHANGELOG.md view
@@ -1,5 +1,20 @@ # Changelog for pgmq-migration +## 0.7.0.0 -- 2026-10-26++### Bug Fixes++- Append `0007-notify-only-registered-queues.sql`: reject physical partitions before+ throttle updates and restrict missing-row fail-open to registered queue names.+ Existing triggers adopt the correction without recreation. Ordinary queues retain+ crash fail-open until an explicit enable/reconcile restores the throttle row.+ Migrations 0001–0006 are unchanged.++### Other Changes++- Ephemeral PostgreSQL roots are created/validated as owner-only (`0700`), existing+ owned roots are tightened, unsafe paths are rejected and registry contents survive.+ ## 0.6.1.1 -- 2026-09-25 Accept the `pg-migrate` 1.2 series alongside 1.1: `pg-migrate`, `pg-migrate-embed` and
+ migrations/0007-notify-only-registered-queues.sql view
@@ -0,0 +1,74 @@+-- Notification fail-open only for a registered queue.+--+-- Migration 0003 made pgmq.notify_queue_listeners() fail open when its throttle+-- row is missing, so crash recovery truncating the UNLOGGED throttle table+-- cannot silently stop delivery. That branch could not tell a row lost to a+-- crash from a missing leaf-name row. PostgreSQL clones a row trigger onto+-- every leaf partition, and inside the clone TG_TABLE_NAME is the partition+-- (q_pq_p0), so the extracted name (pq_p0) normally has no throttle row. On a+-- partitioned queue the fail-open therefore fired on every insert, unthrottled,+-- on a per-partition channel no listener on the documented channel receives.+--+-- The fail-open now requires the extracted name to be a queue registered in+-- pgmq.meta. A relation guard before UPDATE rejects leaf partitions even+-- when their extracted names have alias metadata or throttle rows. Ordinary+-- queues keep crash recovery; a leaf partition, or a+-- mixed-case throttle row the trigger cannot match, behaves as upstream's+-- trigger does: no notification. Partitioned queues receive no insert+-- notifications on the hardened native install. Stock SQL is silent for+-- normal partitioned queues but does not guard manually aliased leaves.+--+-- See docs/design/015-notification-delivery-contract.md and+-- docs/adr/notification-override-lineage-and-partitioned-queue-boundary.md.++CREATE OR REPLACE FUNCTION pgmq.notify_queue_listeners()+RETURNS TRIGGER AS $$+DECLARE+ queue_name_extracted TEXT; -- Queue name extracted from trigger table name+ updated_count INTEGER; -- Number of rows updated (0 or 1)+BEGIN+ -- Metadata names do not prove that this is an ordinary queue relation.+ IF EXISTS (+ SELECT 1 FROM pg_catalog.pg_class c+ WHERE c.oid = TG_RELID AND c.relispartition+ ) THEN+ RETURN NEW;+ END IF;++ queue_name_extracted := substring(TG_TABLE_NAME from 3);++ UPDATE pgmq.notify_insert_throttle+ SET last_notified_at = clock_timestamp()+ WHERE queue_name = queue_name_extracted+ AND (+ throttle_interval_ms = 0 -- No throttling configured+ OR clock_timestamp() - last_notified_at >=+ (throttle_interval_ms * INTERVAL '1 millisecond') -- Throttle interval has elapsed+ );++ -- Check how many rows were updated (will be 0 or 1)+ GET DIAGNOSTICS updated_count = ROW_COUNT;++ IF updated_count > 0 THEN+ PERFORM PG_NOTIFY('pgmq.' || TG_TABLE_NAME || '.' || TG_OP, NULL);+ ELSIF EXISTS (+ SELECT 1 FROM pgmq.meta m+ WHERE m.queue_name = queue_name_extracted+ )+ AND NOT EXISTS (+ SELECT 1 FROM pgmq.notify_insert_throttle nit+ WHERE nit.queue_name = queue_name_extracted+ ) THEN+ -- Fail open only for a queue that could have had a throttle row: the+ -- extracted name is registered in pgmq.meta, so a missing throttle row+ -- can be crash truncation. Both+ -- probes are unique-index lookups and run only when the UPDATE matched+ -- nothing. The row is deliberately NOT re-inserted here (see 0003): the+ -- configured interval is the crash's data loss, and the reconciler+ -- restores it when explicitly invoked, typically at application startup.+ PERFORM PG_NOTIFY('pgmq.' || TG_TABLE_NAME || '.' || TG_OP, NULL);+ END IF;++RETURN NEW;+END;+$$ LANGUAGE plpgsql;
migrations/manifest view
@@ -4,3 +4,4 @@ 0004-upgrade-v1.12.0.sql 0005-upgrade-v1.13.0.sql 0006-preserve-partitioned-reentry-v1.13.0.sql+0007-notify-only-registered-queues.sql
pgmq-migration.cabal view
@@ -1,6 +1,6 @@ cabal-version: 3.4 name: pgmq-migration-version: 0.6.1.1+version: 0.7.0.0 synopsis: PGMQ schema migrations without PostgreSQL extension description: Installs the PGMQ schema into PostgreSQL without requiring the pgmq extension.@@ -75,16 +75,21 @@ type: exitcode-stdio-1.0 hs-source-dirs: test main-is: Main.hs+ other-modules:+ NotifyListener+ RootPermissionsSpec+ build-depends: base >=4.18 && <5, bytestring, containers, directory,- ephemeral-pg >=0.3.1.0,+ ephemeral-pg >=0.3.1 && <0.4, hasql, pg-migrate, pg-migrate-import-hasql-migration,- pgmq-migration >=0.6 && <0.7,+ pgmq-migration >=0.7 && <0.8,+ postgresql-libpq >=0.10.1 && <0.12, tasty ^>=1.5, tasty-hunit ^>=0.10, text,
test/Main.hs view
@@ -4,8 +4,9 @@ module Main (main) where import Control.Concurrent (forkIO, newEmptyMVar, putMVar, takeMVar)-import Control.Exception (SomeException, try)-import Control.Monad (filterM, forM_)+import Control.Exception (SomeException, finally, try)+import Control.Monad (filterM, forM_, unless)+import Data.Bits ((.&.)) import Data.ByteString qualified as ByteString import Data.Foldable (toList) import Data.List.NonEmpty (NonEmpty (..))@@ -56,8 +57,11 @@ migrationNameText, planDescription, )+ -- MigrationDescription's 'migrationId' field would shadow the smart constructor -- of the same name imported above, so it is reached through this alias only.++import Database.PostgreSQL.LibPQ qualified as PQ import Database.PostgreSQL.Migrate.Internal qualified as MigrateInternal import EphemeralPg ( Config (temporaryRoot),@@ -66,6 +70,7 @@ defaultConfig, withCachedConfig, )+import EphemeralPg qualified as Pg import Hasql.Connection qualified as Connection import Hasql.Connection.Settings qualified as Settings import Hasql.Decoders qualified as Decoders@@ -74,6 +79,7 @@ import Hasql.Session qualified as Session import Hasql.Statement (preparable) import Hasql.Statement qualified as Statement+import NotifyListener (awaitNotify, withListener) import Pgmq.Migration qualified as Migration import Pgmq.Migration.History.HasqlMigration ( AlternativeHistoryPolicy (..),@@ -82,8 +88,12 @@ pgmqHasqlMigrationSourceConfig, pgmqHasqlMigrationSourceConfigWithPolicy, )-import System.Directory (createDirectoryIfMissing, doesFileExist)+import RootPermissionsSpec qualified+import System.Directory (doesFileExist) import System.Environment (lookupEnv)+import System.IO.Error (catchIOError, isAlreadyExistsError)+import System.Posix.Directory qualified as Posix+import System.Posix.Files (deviceID, fileID, fileMode, fileOwner, getSymbolicLinkStatus, isDirectory, setFileMode) import System.Posix.User (getEffectiveUserID) import Test.Tasty (TestTree, defaultMain, testGroup) import Test.Tasty.HUnit (assertBool, assertFailure, testCase, (@?=))@@ -103,11 +113,36 @@ uid <- getEffectiveUserID pure ("/tmp/ephpg-pgmq-hs-" <> show uid) +-- | Establish a real effective-uid-owned root without deleting its contents.+ensureOwnerOnlyRoot :: FilePath -> IO ()+ensureOwnerOnlyRoot root = do+ catchIOError (Posix.createDirectory root 0o700) $ \err ->+ unless (isAlreadyExistsError err) (ioError err)+ uid <- getEffectiveUserID+ before <- getSymbolicLinkStatus root+ unless (isDirectory before && fileOwner before == uid) $+ ioError (userError ("Unsafe ephemeral PostgreSQL root: " <> root))+ setFileMode root 0o700+ after <- getSymbolicLinkStatus root+ unless+ ( isDirectory after+ && fileOwner after+ == uid+ && deviceID before+ == deviceID after+ && fileID before+ == fileID after+ && fileMode after+ .&. 0o7777+ == 0o700+ )+ (ioError (userError ("Ephemeral PostgreSQL root changed or is not owner-only: " <> root)))+ -- | Cached-startup configuration pinned to 'ephemeralRoot'. ephemeralConfig :: IO Config ephemeralConfig = do root <- ephemeralRoot- createDirectoryIfMissing True root+ ensureOwnerOnlyRoot root pure defaultConfig {temporaryRoot = Last (Just root)} main :: IO ()@@ -119,16 +154,17 @@ case connResult of Left err -> error $ "Failed to connect: " <> show err Right conn ->- defaultMain (tests connSettings conn)+ defaultMain (tests (Pg.connectionString db) connSettings conn) case result of Left startErr -> error $ "Failed to start temp database: " <> show startErr Right () -> pure () -tests :: Settings.Settings -> Connection.Connection -> TestTree-tests settings conn =+tests :: Text -> Settings.Settings -> Connection.Connection -> TestTree+tests conninfo settings conn = testGroup "pgmq-migration"- [ testGroup+ [ RootPermissionsSpec.tests ensureOwnerOnlyRoot,+ testGroup "native definition" [ testCase "immutable history and upstream payload provenance" (testNativePayload conn), testCase "component pgmq lists the ledger in order and has no dependencies" testNativeComponent@@ -137,7 +173,8 @@ "native runner" [ testCase "fresh install applies once and is idempotent" (testNativeRunner settings conn), testCase "1.12 and 1.13 catalog convergence" (testConvergence conn),- testCase "populated partition upgrades and spill recovery" (testPartitions settings conn)+ testCase "populated partition upgrades and spill recovery" (testPartitions settings conn),+ testCase "existing triggers adopt notification correction on populated upgrade" (testNotifyUpgrade conninfo settings conn) ], testGroup "history import"@@ -155,7 +192,8 @@ forM_ [ ("0001-install-v1.11.0.sql", "faa9b8800005f80fbdf6a33d071183d2"), ("0002-schema-management-comment.sql", "f3010777d47ec6d4f7d92a179bcf09b8"),- ("0003-notify-crash-safety-and-locking.sql", "e0a66c96bd8811607440dbca0cb302ea")+ ("0003-notify-crash-safety-and-locking.sql", "e0a66c96bd8811607440dbca0cb302ea"),+ ("0007-notify-only-registered-queues.sql", "2892dc6622062d1530a3fa29c17039bb") ] $ \(name, digest) -> do bytes <- readMigration name@@ -183,7 +221,7 @@ testConvergence conn = forM_ [False, True] $ \latest -> do resetDb conn names <- nativeMigrationNames- forM_ (take (if latest then 6 else 4) names) $ \name ->+ forM_ (take (if latest then length names else 4) names) $ \name -> readMigration (T.unpack name <> ".sql") >>= runSql conn . Text.decodeUtf8 upgraded <- schemaSnapshot conn resetDb conn@@ -339,6 +377,60 @@ resetDb conn runSql conn "DROP EXTENSION pg_partman CASCADE" +-- | Install six entries, preserve deployed trigger OIDs/data/settings, then apply 0007.+testNotifyUpgrade :: Text -> Settings.Settings -> Connection.Connection -> IO ()+testNotifyUpgrade conninfo settings conn = do+ required <- (== Just "1") <$> lookupEnv "PGMQ_REQUIRE_PARTMAN"+ available <- queryText conn "SELECT EXISTS (SELECT FROM pg_available_extensions WHERE name='pg_partman')::text"+ if available /= "true"+ then if required then assertFailure "pg_partman required" else putStrLn "SKIPPED: notification upgrade requires pg_partman"+ else flip finally (resetDb conn >> runSql conn "DROP EXTENSION IF EXISTS pg_partman CASCADE") $ do+ resetDb conn+ runSql conn "CREATE SCHEMA IF NOT EXISTS partman; CREATE EXTENSION IF NOT EXISTS pg_partman SCHEMA partman"+ installPrefix settings 6+ runSql conn "SELECT pgmq.create('notify_ordinary'); SELECT pgmq.create_partitioned('notify_partition','10','100'); SELECT pgmq.enable_notify_insert('notify_ordinary',0); SELECT pgmq.enable_notify_insert('notify_partition',0)"+ withListener conninfo "pgmq.q_notify_partition_p0.INSERT" $ \listener -> do+ runSql conn "SELECT pgmq.send('notify_partition','{\"before\":true}')"+ beforeNotify <- awaitNotify listener 20+ fmap PQ.notifyRelname beforeNotify @?= Just "pgmq.q_notify_partition_p0.INSERT"+ runSql conn "SELECT pgmq.send('notify_ordinary','{\"before\":true}')"+ before <- queryText conn notifyUpgradeSnapshot+ plan <- nativePlan+ report <- runMigrationPlan defaultRunOptions settings plan >>= either (assertFailure . show) pure+ map outcome (toList (results report)) @?= replicate 6 AlreadyApplied <> [AppliedNow]+ queryText conn notifyUpgradeSnapshot >>= (@?= before)+ -- No enable/reconcile occurs: every parent and cloned trigger retains its OID.+ runListen listener "pgmq.q_notify_partition.INSERT"+ partitionResult <- PQ.exec listener "SELECT 'pgmq.' || c.relname || '.INSERT' FROM pg_inherits i JOIN pg_class c ON c.oid=i.inhrelid WHERE i.inhparent='pgmq.q_notify_partition'::regclass"+ case partitionResult of+ Nothing -> assertFailure "partition channel query failed"+ Just res -> do+ PQ.resultStatus res >>= (@?= PQ.TuplesOk)+ rows <- PQ.ntuples res+ forM_ [0 .. rows - 1] $ \row -> do+ value <- PQ.getvalue res row 0+ maybe (assertFailure "null partition channel") (runListen listener . Text.decodeUtf8) value+ forM_ [1 :: Int .. 3] $ \_ -> runSql conn "SELECT pgmq.send('notify_partition','{\"after\":true}')"+ silent <- awaitNotify listener 10+ fmap PQ.notifyRelname silent @?= Nothing+ assertSql conn "(SELECT count(*)=4 FROM pgmq.q_notify_partition) AND (SELECT last_notified_at=to_timestamp(0) FROM pgmq.notify_insert_throttle WHERE queue_name='notify_partition')"+ withListener conninfo "pgmq.q_notify_ordinary.INSERT" $ \listener -> do+ runSql conn "SELECT pgmq.send('notify_ordinary','{\"after\":true}')"+ ordinary <- awaitNotify listener 20+ fmap PQ.notifyRelname ordinary @?= Just "pgmq.q_notify_ordinary.INSERT"+ plan <- nativePlan+ again <- runMigrationPlan defaultRunOptions settings plan >>= either (assertFailure . show) pure+ map outcome (toList (results again)) @?= replicate 7 AlreadyApplied+ where+ runListen listener channel = do+ result <- PQ.exec listener (Text.encodeUtf8 ("LISTEN \"" <> channel <> "\""))+ case result of+ Nothing -> assertFailure "LISTEN failed"+ Just res -> PQ.resultStatus res >>= (@?= PQ.CommandOk)++notifyUpgradeSnapshot :: Text+notifyUpgradeSnapshot = "SELECT jsonb_build_object('data',(SELECT jsonb_agg(to_jsonb(m) ORDER BY source,msg_id) FROM (SELECT 'ordinary' source,msg_id,message FROM pgmq.q_notify_ordinary UNION ALL SELECT 'partition',msg_id,message FROM pgmq.q_notify_partition) m),'triggers',(SELECT jsonb_agg(jsonb_build_array(oid::text,tgrelid::text,tgfoid::text,tgparentid::text) ORDER BY oid) FROM pg_trigger WHERE tgname='trigger_notify_queue_insert_listeners'),'throttles',(SELECT jsonb_agg(to_jsonb(t) ORDER BY queue_name) FROM pgmq.notify_insert_throttle t))::text"+ preservedDataSql :: Text preservedDataSql = "SELECT jsonb_agg(to_jsonb(m) ORDER BY source,msg_id)::text FROM (SELECT 'numeric' source,msg_id,message FROM pgmq.q_numeric_old UNION ALL SELECT 'archive',msg_id,message FROM pgmq.a_numeric_old UNION ALL SELECT 'time',msg_id,message FROM pgmq.q_time_old UNION ALL SELECT 'ordinary',msg_id,message FROM pgmq.q_ordinary_old) m" @@ -363,7 +455,8 @@ "0003-notify-crash-safety-and-locking", "0004-upgrade-v1.12.0", "0005-upgrade-v1.13.0",- "0006-preserve-partitioned-reentry-v1.13.0"+ "0006-preserve-partitioned-reentry-v1.13.0",+ "0007-notify-only-registered-queues" ] actual -> assertFailure ("unexpected native PGMQ plan: " <> show actual)
+ test/NotifyListener.hs view
@@ -0,0 +1,48 @@+{-# LANGUAGE OverloadedStrings #-}++-- | Checked LISTEN transport for deployed-upgrade acceptance.+module NotifyListener (withListener, awaitNotify) where++import Control.Concurrent (threadDelay)+import Control.Exception (bracket)+import Control.Monad (unless)+import Data.Text (Text)+import Data.Text qualified as T+import Data.Text.Encoding qualified as TE+import Database.PostgreSQL.LibPQ qualified as LibPQ+import Test.Tasty.HUnit (assertFailure, (@?=))++withListener :: Text -> Text -> (LibPQ.Connection -> IO a) -> IO a+withListener conninfo channel action =+ bracket (LibPQ.connectdb (TE.encodeUtf8 conninfo)) LibPQ.finish $ \conn -> do+ connStatus <- LibPQ.status conn+ unless (connStatus == LibPQ.ConnectionOk) $ do+ err <- LibPQ.errorMessage conn+ assertFailure $ "libpq connection failed: " <> show err+ -- The channel contains dots, so LISTEN needs the identifier double-quoted.+ result <- LibPQ.exec conn (TE.encodeUtf8 ("LISTEN " <> quoteIdentifier channel))+ case result of+ Nothing -> assertFailure "LISTEN returned no result"+ Just res -> do+ execStatus <- LibPQ.resultStatus res+ unless (execStatus == LibPQ.CommandOk) $+ assertFailure ("LISTEN failed with " <> show execStatus)+ action conn++quoteIdentifier :: Text -> Text+quoteIdentifier ident = "\"" <> T.replace "\"" "\"\"" ident <> "\""++-- | Poll for a notification, 100 ms per attempt.+awaitNotify :: LibPQ.Connection -> Int -> IO (Maybe LibPQ.Notify)+awaitNotify conn attempts+ | attempts <= 0 = pure Nothing+ | otherwise = do+ ok <- LibPQ.consumeInput conn+ unless ok (assertFailure "libpq consumeInput failed")+ LibPQ.status conn >>= (@?= LibPQ.ConnectionOk)+ pending <- LibPQ.notifies conn+ case pending of+ Just n -> pure (Just n)+ Nothing -> do+ threadDelay 100_000+ awaitNotify conn (attempts - 1)
+ test/RootPermissionsSpec.hs view
@@ -0,0 +1,63 @@+-- | Check each suite's local root helper using private disposable fixtures.+module RootPermissionsSpec (tests) where++import Control.Exception (IOException, bracket, try)+import Data.Bits ((.&.))+import System.Directory (getTemporaryDirectory, removePathForcibly)+import System.Posix.Directory qualified as Posix+import System.Posix.Files (createSymbolicLink, fileMode, getSymbolicLinkStatus, isDirectory, setFileMode)+import System.Posix.Temp (mkdtemp)+import Test.Tasty (TestTree, testGroup)+import Test.Tasty.HUnit (assertBool, assertFailure, testCase, (@?=))++tests :: (FilePath -> IO ()) -> TestTree+tests ensure =+ testGroup+ "RootPermissionsSpec"+ [ testCase "missing root is a real 0700 directory" $ private $ \parent -> do+ let root = parent <> "/root"+ ensure root+ status <- getSymbolicLinkStatus root+ assertBool "real directory" (isDirectory status)+ fileMode status .&. 0o7777 @?= 0o700,+ testCase "owned 0755 directory is tightened without losing contents" $ private $ \parent -> do+ let root = parent <> "/root"+ Posix.createDirectory root 0o755+ setFileMode root 0o755+ writeFile (root <> "/sentinel") "preserve"+ ensure root+ mode root >>= (@?= 0o700)+ readFile (root <> "/sentinel") >>= (@?= "preserve"),+ testCase "repeated call preserves contents and permissions" $ private $ \parent -> do+ let root = parent <> "/root"+ ensure root+ writeFile (root <> "/sentinel") "preserve"+ ensure root+ mode root >>= (@?= 0o700)+ readFile (root <> "/sentinel") >>= (@?= "preserve"),+ testCase "regular file is rejected unchanged" $ private $ \parent -> do+ let root = parent <> "/root"+ writeFile root "preserve"+ setFileMode root 0o644+ rejected (ensure root)+ mode root >>= (@?= 0o644)+ readFile root >>= (@?= "preserve"),+ testCase "symlink is rejected without chmod of its target" $ private $ \parent -> do+ let root = parent <> "/root"+ target = parent <> "/target"+ Posix.createDirectory target 0o755+ setFileMode target 0o755+ writeFile (target <> "/sentinel") "preserve"+ createSymbolicLink target root+ rejected (ensure root)+ mode target >>= (@?= 0o755)+ readFile (target <> "/sentinel") >>= (@?= "preserve")+ ]+ where+ mode path = (.&. 0o7777) . fileMode <$> getSymbolicLinkStatus path+ rejected action = do+ result <- try action :: IO (Either IOException ())+ case result of+ Left _ -> pure ()+ Right () -> assertFailure "unsafe root accepted"+ private action = bracket (getTemporaryDirectory >>= \tmp -> mkdtemp (tmp <> "/pgmq-root-test-")) removePathForcibly action