From 13eef4cfd8b58192af6a7a2893e38be59da7b217 Mon Sep 17 00:00:00 2001 From: max Date: Wed, 7 Oct 2026 12:25:24 +0300 Subject: [PATCH 1/4] Do not recreate the notify trigger when a consumer starts Consumer::consume() called enable_notify_insert() on every start, which drops and recreates the trigger on the queue table. That takes a table lock: a (re)starting worker waits for every open transaction that inserted into the queue, blocks new inserts meanwhile, and two workers of one queue starting together can deadlock. Add ensureNotifyInsert()/isNotifyInsertEnabled(): the consumer now only checks the catalog and enables notifications when they are missing or use another throttle. --- src/ConsumeConfig.php | 2 +- src/Consumer.php | 2 +- src/Queue.php | 21 +++++++++++++ src/queue.fn.php | 71 ++++++++++++++++++++++++++++++++++++++++++- tests/PgmqTest.php | 52 +++++++++++++++++++++++++++++++ 5 files changed, 145 insertions(+), 3 deletions(-) diff --git a/src/ConsumeConfig.php b/src/ConsumeConfig.php index f02604b..e7f08cd 100644 --- a/src/ConsumeConfig.php +++ b/src/ConsumeConfig.php @@ -19,7 +19,7 @@ /** * @param non-empty-string $queue * @param positive-int $batch - * @param bool $listenForInserts will call {@see enableNotifyInsert} and will listen for notifications from the channel, which significantly optimizes the number of requests to the Postgres server for new messages. It is recommended to enable it. + * @param bool $listenForInserts will call {@see ensureNotifyInsert} and will listen for notifications from the channel, which significantly optimizes the number of requests to the Postgres server for new messages. It is recommended to enable it. */ public function __construct( public string $queue, diff --git a/src/Consumer.php b/src/Consumer.php index a6bd5de..99f00b4 100644 --- a/src/Consumer.php +++ b/src/Consumer.php @@ -48,7 +48,7 @@ public function consume( $watchers[] = $timeoutWatcher; if ($config->listenForInserts) { - $channelName = $queue->enableNotifyInsert(); + $channelName = $queue->ensureNotifyInsert(); $watchers[] = new Internal\ChannelWatcher( $polls, diff --git a/src/Queue.php b/src/Queue.php index fba37af..0307f68 100644 --- a/src/Queue.php +++ b/src/Queue.php @@ -295,6 +295,27 @@ public function enableNotifyInsert(?TimeSpan $throttleInterval = null): string ); } + /** + * @return non-empty-string + */ + public function ensureNotifyInsert(?TimeSpan $throttleInterval = null): string + { + return ensureNotifyInsert( + pg: $this->pg, + queue: $this->name, + throttleInterval: $throttleInterval, + ); + } + + public function isNotifyInsertEnabled(?TimeSpan $throttleInterval = null): bool + { + return isNotifyInsertEnabled( + pg: $this->pg, + queue: $this->name, + throttleInterval: $throttleInterval, + ); + } + public function disableNotifyInsert(): void { disableNotifyInsert( diff --git a/src/queue.fn.php b/src/queue.fn.php index 545e79f..a666fad 100644 --- a/src/queue.fn.php +++ b/src/queue.fn.php @@ -461,12 +461,81 @@ function enableNotifyInsert( ): string { $pg->execute('SELECT pgmq.enable_notify_insert(:queue_name, :throttle_interval_ms)', [ 'queue_name' => $queue, - 'throttle_interval_ms' => ($throttleInterval ?? TimeSpan::fromMilliseconds(30))->toMilliseconds(), + 'throttle_interval_ms' => notifyThrottleMs($throttleInterval), ]); return channelName($queue); } +/** + * Enables insert notifications only when they are not already enabled with the same throttle interval. + * Unlike {@see enableNotifyInsert()}, which always recreates the trigger, an existing setup is left untouched: + * recreating the trigger takes a table lock that waits for every open transaction inserting into the queue + * and blocks new inserts meanwhile. + * + * @api + * @param non-empty-string $queue + * @return non-empty-string + */ +function ensureNotifyInsert( + PostgresLink $pg, + string $queue, + ?TimeSpan $throttleInterval = null, +): string { + if (!isNotifyInsertEnabled($pg, $queue, $throttleInterval)) { + enableNotifyInsert($pg, $queue, $throttleInterval); + } + + return channelName($queue); +} + +/** + * Reads the catalog only: takes no lock on the queue table. + * + * @api + * @param non-empty-string $queue + */ +function isNotifyInsertEnabled( + PostgresLink $pg, + string $queue, + ?TimeSpan $throttleInterval = null, +): bool { + /** @var ?array{enabled: bool} $row */ + $row = $pg + ->execute( + <<<'SQL' + SELECT EXISTS ( + SELECT 1 + FROM pgmq.notify_insert_throttle t + WHERE t.queue_name = :queue_name AND t.throttle_interval_ms = :throttle_interval_ms + ) AND EXISTS ( + SELECT 1 + FROM pg_trigger tg + JOIN pg_class c ON c.oid = tg.tgrelid + JOIN pg_namespace n ON n.oid = c.relnamespace + WHERE n.nspname = 'pgmq' + AND c.relname = pgmq.format_table_name(:queue_name, 'q') + AND tg.tgname = 'trigger_notify_queue_insert_listeners' + ) AS enabled + SQL, + [ + 'queue_name' => $queue, + 'throttle_interval_ms' => notifyThrottleMs($throttleInterval), + ], + ) + ->fetchRow(); + + return $row !== null && $row['enabled']; +} + +/** + * @internal + */ +function notifyThrottleMs(?TimeSpan $throttleInterval): int +{ + return ($throttleInterval ?? TimeSpan::fromMilliseconds(30))->toMilliseconds(); +} + /** * @api * @param non-empty-string $queue diff --git a/tests/PgmqTest.php b/tests/PgmqTest.php index ce2f1f0..5ce2af6 100644 --- a/tests/PgmqTest.php +++ b/tests/PgmqTest.php @@ -8,10 +8,12 @@ use Amp\Postgres\PostgresConnection; use Amp\Postgres\PostgresConnectionPool; use Amp\Postgres\PostgresQueryError; +use Amp\TimeoutCancellation; use PHPUnit\Framework\Attributes\CoversClass; use PHPUnit\Framework\Attributes\DoesNotPerformAssertions; use PHPUnit\Framework\TestCase; use Thesis\Time\TimeSpan; +use function Amp\async; use function Amp\delay; #[CoversClass(Queue::class)] @@ -276,6 +278,56 @@ public function testEnableDisableNotifies(): void $queue->disableNotifyInsert(); } + public function testConsumerStartDoesNotWaitForProducerTransactions(): void + { + $queue = createQueue($this->pg, $this->randomQueueName()); + $queue->enableNotifyInsert(); + + // A producer is in the middle of a transaction that has already inserted into the queue. + $producer = $this->pg->beginTransaction(); + send($producer, $queue->name, new SendMessage(self::TESTING_MESSAGE)); + + $pg = $this->pg; + + try { + // A worker (re)starts: notifications are already set up, so it must not touch the trigger. Recreating it + // takes a table lock that waits for every open producer transaction and blocks new inserts meanwhile. + /** @var ConsumeContext $context */ + $context = async(static fn(): ConsumeContext => createConsumer($pg)->consume( + static function (): void {}, + new ConsumeConfig($queue->name), + ))->await(new TimeoutCancellation(1)); + + $context->stop(); + $context->awaitCompletion(); + } finally { + $producer->rollback(); + } + + self::assertTrue($queue->isNotifyInsertEnabled()); + } + + public function testEnsureNotifyInsertKeepsExistingSetupAndEnablesMissingOne(): void + { + $queue = createQueue($this->pg, $this->randomQueueName()); + + self::assertFalse($queue->isNotifyInsertEnabled()); + self::assertSame(channelName($queue->name), $queue->ensureNotifyInsert()); + self::assertTrue($queue->isNotifyInsertEnabled()); + + // Already enabled with the same throttle: nothing to do. + self::assertSame(channelName($queue->name), $queue->ensureNotifyInsert()); + self::assertTrue($queue->isNotifyInsertEnabled()); + + // A different throttle is a different setup: it is applied. + $queue->ensureNotifyInsert(TimeSpan::fromSeconds(1)); + self::assertTrue($queue->isNotifyInsertEnabled(TimeSpan::fromSeconds(1))); + self::assertFalse($queue->isNotifyInsertEnabled()); + + $queue->disableNotifyInsert(); + self::assertFalse($queue->isNotifyInsertEnabled(TimeSpan::fromSeconds(1))); + } + public function testInvalidConsumerConfiguration(): void { $queue = createQueue($this->pg, $this->randomQueueName()); From af05f8326c47fb66c10ee0717c5946524ed932ed Mon Sep 17 00:00:00 2001 From: max Date: Wed, 7 Oct 2026 14:03:39 +0300 Subject: [PATCH 2/4] ensureNotifyInsert(): serialize under the queue lock, check trigger validity and the throttle row - check and change run in one transaction under pgmq.acquire_queue_lock(), so consumers starting together no longer race into enable_notify_insert() (concurrent cold starts failed with 'trigger ... already exists'); - the trigger counts as valid only when enabled and calling notify_queue_listeners(); - a missing throttle row (UNLOGGED: emptied by crash recovery) re-enables notifications; - a different throttle interval uses update_notify_insert(), without DDL. --- src/Internal/NotifyInsertState.php | 16 +++++ src/queue.fn.php | 100 +++++++++++++++++++++-------- 2 files changed, 90 insertions(+), 26 deletions(-) create mode 100644 src/Internal/NotifyInsertState.php diff --git a/src/Internal/NotifyInsertState.php b/src/Internal/NotifyInsertState.php new file mode 100644 index 0000000..e8b8da8 --- /dev/null +++ b/src/Internal/NotifyInsertState.php @@ -0,0 +1,16 @@ +beginTransaction(); + + try { + $tx->execute('SELECT pgmq.acquire_queue_lock(:queue_name)', ['queue_name' => $queue]); + $state = notifyInsertState($tx, $queue); + + if (!$state->triggerValid || $state->throttleIntervalMs === null) { + $tx->execute('SELECT pgmq.enable_notify_insert(:queue_name, :throttle_interval_ms)', [ + 'queue_name' => $queue, + 'throttle_interval_ms' => $throttleMs, + ]); + } elseif ($state->throttleIntervalMs !== $throttleMs) { + $tx->execute('SELECT pgmq.update_notify_insert(:queue_name, :throttle_interval_ms)', [ + 'queue_name' => $queue, + 'throttle_interval_ms' => $throttleMs, + ]); + } + + $tx->commit(); + } catch (\Throwable $e) { + if ($tx->isActive()) { + $tx->rollback(); + } + + throw $e; } return channelName($queue); } /** - * Reads the catalog only: takes no lock on the queue table. + * Whether insert notifications work for the queue with this throttle interval: the trigger exists, is enabled and + * calls `pgmq.notify_queue_listeners()`, and the throttle row is in place. Reads the catalog only. * * @api * @param non-empty-string $queue @@ -500,32 +533,47 @@ function isNotifyInsertEnabled( string $queue, ?TimeSpan $throttleInterval = null, ): bool { - /** @var ?array{enabled: bool} $row */ + $state = notifyInsertState($pg, $queue); + + return $state->triggerValid && $state->throttleIntervalMs === notifyThrottleMs($throttleInterval); +} + +/** + * @internal + * @param non-empty-string $queue + */ +function notifyInsertState(PostgresLink $pg, string $queue): Internal\NotifyInsertState +{ + /** @var array{trigger_valid: bool, throttle_interval_ms: ?int} $row */ $row = $pg ->execute( <<<'SQL' - SELECT EXISTS ( - SELECT 1 - FROM pgmq.notify_insert_throttle t - WHERE t.queue_name = :queue_name AND t.throttle_interval_ms = :throttle_interval_ms - ) AND EXISTS ( - SELECT 1 - FROM pg_trigger tg - JOIN pg_class c ON c.oid = tg.tgrelid - JOIN pg_namespace n ON n.oid = c.relnamespace - WHERE n.nspname = 'pgmq' - AND c.relname = pgmq.format_table_name(:queue_name, 'q') - AND tg.tgname = 'trigger_notify_queue_insert_listeners' - ) AS enabled + SELECT + EXISTS ( + SELECT 1 + FROM pg_trigger tg + JOIN pg_class c ON c.oid = tg.tgrelid + JOIN pg_namespace n ON n.oid = c.relnamespace + WHERE n.nspname = 'pgmq' + AND c.relname = pgmq.format_table_name(:queue_name, 'q') + AND tg.tgname = 'trigger_notify_queue_insert_listeners' + AND tg.tgenabled IN ('O', 'A') + AND tg.tgfoid = 'pgmq.notify_queue_listeners()'::regprocedure + ) AS trigger_valid, + ( + SELECT t.throttle_interval_ms + FROM pgmq.notify_insert_throttle t + WHERE t.queue_name = :queue_name + ) AS throttle_interval_ms SQL, - [ - 'queue_name' => $queue, - 'throttle_interval_ms' => notifyThrottleMs($throttleInterval), - ], + ['queue_name' => $queue], ) ->fetchRow(); - return $row !== null && $row['enabled']; + return new Internal\NotifyInsertState( + triggerValid: $row['trigger_valid'], + throttleIntervalMs: $row['throttle_interval_ms'], + ); } /** From b33e0928d3fc1abc140a06ea27a84cd381d2287c Mon Sep 17 00:00:00 2001 From: max Date: Wed, 7 Oct 2026 14:22:53 +0300 Subject: [PATCH 3/4] Test ensureNotifyInsert() against concurrent cold starts and broken setups - 4 consumers starting at once on a queue without notifications create the trigger exactly once (counted with an event trigger); without the queue lock they raced and failed with 'trigger ... already exists'; - a lost throttle row (crash recovery), a disabled trigger and a trigger calling another function are restored; - a different throttle interval is applied without recreating the trigger. --- tests/PgmqTest.php | 127 +++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 127 insertions(+) diff --git a/tests/PgmqTest.php b/tests/PgmqTest.php index 5ce2af6..f6130b5 100644 --- a/tests/PgmqTest.php +++ b/tests/PgmqTest.php @@ -328,6 +328,97 @@ public function testEnsureNotifyInsertKeepsExistingSetupAndEnablesMissingOne(): self::assertFalse($queue->isNotifyInsertEnabled(TimeSpan::fromSeconds(1))); } + public function testConcurrentColdStartsCreateTheTriggerOnce(): void + { + $queue = createQueue($this->pg, $this->randomQueueName()); // notifications not set up yet + + // A producer keeps an insert into the queue open, so that every consumer start has to wait for the table lock + // and the starts are guaranteed to overlap. + $producer = $this->pg->beginTransaction(); + send($producer, $queue->name, new SendMessage(self::TESTING_MESSAGE)); + + $dsn = (string) getenv('THESIS_PGMQ_DSN'); + $creations = $this->countTriggerCreations(static function () use ($dsn, $queue, $producer): void { + $starts = []; + + for ($i = 0; $i < 4; ++$i) { + // Each consumer has its own pool, as separate worker processes do (one pool cannot LISTEN twice). + $pool = new PostgresConnectionPool(PostgresConfig::fromString($dsn)); + $starts[] = async(static fn(): ConsumeContext => createConsumer($pool)->consume( + static function (): void {}, + new ConsumeConfig($queue->name), + )); + } + + delay(0.5); + $producer->rollback(); + + foreach ($starts as $start) { + /** @var ConsumeContext $context */ + $context = $start->await(new TimeoutCancellation(5)); + $context->stop(); + $context->awaitCompletion(); + } + }, $queue->name); + + self::assertSame(1, $creations, 'concurrent starts must not race into enable_notify_insert()'); + self::assertTrue($queue->isNotifyInsertEnabled()); + } + + public function testEnsureNotifyInsertRestoresLostThrottleRow(): void + { + $queue = createQueue($this->pg, $this->randomQueueName()); + $queue->enableNotifyInsert(); + + // pgmq.notify_insert_throttle is UNLOGGED: crash recovery empties it while the trigger survives. + $this->pg->execute('DELETE FROM pgmq.notify_insert_throttle WHERE queue_name = :queue_name', ['queue_name' => $queue->name]); + self::assertFalse($queue->isNotifyInsertEnabled()); + + $queue->ensureNotifyInsert(); + + self::assertTrue($queue->isNotifyInsertEnabled()); + } + + public function testEnsureNotifyInsertRecreatesDisabledTrigger(): void + { + $queue = createQueue($this->pg, $this->randomQueueName()); + $queue->enableNotifyInsert(); + $this->pg->query(\sprintf('ALTER TABLE pgmq.q_%s DISABLE TRIGGER trigger_notify_queue_insert_listeners', $queue->name)); + self::assertFalse($queue->isNotifyInsertEnabled()); + + $queue->ensureNotifyInsert(); + + self::assertTrue($queue->isNotifyInsertEnabled()); + } + + public function testEnsureNotifyInsertRecreatesTriggerCallingAnotherFunction(): void + { + $queue = createQueue($this->pg, $this->randomQueueName()); + $queue->enableNotifyInsert(); + $this->pg->query('CREATE OR REPLACE FUNCTION public.thesis_test_noop_trigger() RETURNS trigger LANGUAGE plpgsql AS $$ BEGIN RETURN NEW; END $$'); + $this->pg->query(\sprintf('DROP TRIGGER trigger_notify_queue_insert_listeners ON pgmq.q_%s', $queue->name)); + $this->pg->query(\sprintf( + 'CREATE CONSTRAINT TRIGGER trigger_notify_queue_insert_listeners AFTER INSERT ON pgmq.q_%s DEFERRABLE FOR EACH ROW EXECUTE PROCEDURE public.thesis_test_noop_trigger()', + $queue->name, + )); + self::assertFalse($queue->isNotifyInsertEnabled()); + + $queue->ensureNotifyInsert(); + + self::assertTrue($queue->isNotifyInsertEnabled()); + } + + public function testEnsureNotifyInsertChangesThrottleWithoutRecreatingTrigger(): void + { + $queue = createQueue($this->pg, $this->randomQueueName()); + $queue->enableNotifyInsert(TimeSpan::fromSeconds(1)); + + $creations = $this->countTriggerCreations(static fn() => $queue->ensureNotifyInsert(), $queue->name); + + self::assertSame(0, $creations); + self::assertTrue($queue->isNotifyInsertEnabled()); + } + public function testInvalidConsumerConfiguration(): void { $queue = createQueue($this->pg, $this->randomQueueName()); @@ -689,6 +780,42 @@ public function testValidateTopicPattern(): void validateTopicPattern($this->pg, 'logs.**'); } + /** + * Runs $action and returns how many times the notify trigger of the queue was created meanwhile (event trigger). + * + * @param callable(): mixed $action + * @param non-empty-string $queue + */ + private function countTriggerCreations(callable $action, string $queue): int + { + $this->pg->query('CREATE TABLE IF NOT EXISTS public.thesis_test_ddl (object_identity text)'); + $this->pg->query('TRUNCATE public.thesis_test_ddl'); + $this->pg->query(<<<'SQL' + CREATE OR REPLACE FUNCTION public.thesis_test_log_ddl() RETURNS event_trigger LANGUAGE plpgsql AS $$ + DECLARE r record; + BEGIN + FOR r IN SELECT * FROM pg_event_trigger_ddl_commands() LOOP + INSERT INTO public.thesis_test_ddl (object_identity) VALUES (r.object_identity); + END LOOP; + END $$ + SQL); + $this->pg->query('DROP EVENT TRIGGER IF EXISTS thesis_test_ddl'); + $this->pg->query("CREATE EVENT TRIGGER thesis_test_ddl ON ddl_command_end WHEN TAG IN ('CREATE TRIGGER') EXECUTE FUNCTION public.thesis_test_log_ddl()"); + + try { + $action(); + } finally { + $this->pg->query('DROP EVENT TRIGGER IF EXISTS thesis_test_ddl'); + } + + /** @var array{n: int} $row */ + $row = $this->pg + ->execute('SELECT count(*)::int AS n FROM public.thesis_test_ddl WHERE object_identity LIKE :pattern', ['pattern' => '% on pgmq.q_' . $queue]) + ->fetchRow(); + + return $row['n']; + } + /** * @return non-empty-string */ From 8bc6f61e78a9da6bd3da51a898e131ba2593a96d Mon Sep 17 00:00:00 2001 From: max Date: Wed, 7 Oct 2026 14:50:41 +0300 Subject: [PATCH 4/4] ensureNotifyInsert(): one statement instead of an own transaction Lock, check and change now run in a single DO block, so ensureNotifyInsert() needs no transaction of its own: it works on any PostgresLink (a pool, a connection, or an open transaction, where the queue lock is held until that transaction ends) and costs one round trip instead of five. The trigger-validity condition is defined once and shared with isNotifyInsertEnabled(). --- src/Internal/NotifyInsertState.php | 16 ---- src/queue.fn.php | 131 +++++++++++++++-------------- tests/PgmqTest.php | 12 +++ 3 files changed, 81 insertions(+), 78 deletions(-) delete mode 100644 src/Internal/NotifyInsertState.php diff --git a/src/Internal/NotifyInsertState.php b/src/Internal/NotifyInsertState.php deleted file mode 100644 index e8b8da8..0000000 --- a/src/Internal/NotifyInsertState.php +++ /dev/null @@ -1,16 +0,0 @@ -beginTransaction(); - - try { - $tx->execute('SELECT pgmq.acquire_queue_lock(:queue_name)', ['queue_name' => $queue]); - $state = notifyInsertState($tx, $queue); - - if (!$state->triggerValid || $state->throttleIntervalMs === null) { - $tx->execute('SELECT pgmq.enable_notify_insert(:queue_name, :throttle_interval_ms)', [ - 'queue_name' => $queue, - 'throttle_interval_ms' => $throttleMs, - ]); - } elseif ($state->throttleIntervalMs !== $throttleMs) { - $tx->execute('SELECT pgmq.update_notify_insert(:queue_name, :throttle_interval_ms)', [ - 'queue_name' => $queue, - 'throttle_interval_ms' => $throttleMs, - ]); - } - - $tx->commit(); - } catch (\Throwable $e) { - if ($tx->isActive()) { - $tx->rollback(); - } - - throw $e; - } + $pg->query(\sprintf( + <<<'SQL' + DO $ensure$ + DECLARE + v_queue text := %s; + v_throttle_ms integer := %d; + v_current_ms integer; + BEGIN + PERFORM pgmq.acquire_queue_lock(v_queue); + + SELECT t.throttle_interval_ms INTO v_current_ms + FROM pgmq.notify_insert_throttle t + WHERE t.queue_name = v_queue; + + IF NOT (%s) OR v_current_ms IS NULL THEN + PERFORM pgmq.enable_notify_insert(v_queue, v_throttle_ms); + ELSIF v_current_ms <> v_throttle_ms THEN + PERFORM pgmq.update_notify_insert(v_queue, v_throttle_ms); + END IF; + END + $ensure$ + SQL, + $pg->quoteLiteral($queue), + notifyThrottleMs($throttleInterval), + notifyTriggerValidSql('v_queue'), + )); return channelName($queue); } @@ -533,46 +533,53 @@ function isNotifyInsertEnabled( string $queue, ?TimeSpan $throttleInterval = null, ): bool { - $state = notifyInsertState($pg, $queue); - - return $state->triggerValid && $state->throttleIntervalMs === notifyThrottleMs($throttleInterval); -} - -/** - * @internal - * @param non-empty-string $queue - */ -function notifyInsertState(PostgresLink $pg, string $queue): Internal\NotifyInsertState -{ - /** @var array{trigger_valid: bool, throttle_interval_ms: ?int} $row */ + /** @var array{enabled: bool} $row */ $row = $pg ->execute( - <<<'SQL' - SELECT - EXISTS ( + \sprintf( + <<<'SQL' + SELECT (%s) AND EXISTS ( SELECT 1 - FROM pg_trigger tg - JOIN pg_class c ON c.oid = tg.tgrelid - JOIN pg_namespace n ON n.oid = c.relnamespace - WHERE n.nspname = 'pgmq' - AND c.relname = pgmq.format_table_name(:queue_name, 'q') - AND tg.tgname = 'trigger_notify_queue_insert_listeners' - AND tg.tgenabled IN ('O', 'A') - AND tg.tgfoid = 'pgmq.notify_queue_listeners()'::regprocedure - ) AS trigger_valid, - ( - SELECT t.throttle_interval_ms FROM pgmq.notify_insert_throttle t - WHERE t.queue_name = :queue_name - ) AS throttle_interval_ms - SQL, - ['queue_name' => $queue], + WHERE t.queue_name = :queue_name AND t.throttle_interval_ms = :throttle_interval_ms + ) AS enabled + SQL, + notifyTriggerValidSql(':queue_name'), + ), + [ + 'queue_name' => $queue, + 'throttle_interval_ms' => notifyThrottleMs($throttleInterval), + ], ) ->fetchRow(); - return new Internal\NotifyInsertState( - triggerValid: $row['trigger_valid'], - throttleIntervalMs: $row['throttle_interval_ms'], + return $row['enabled']; +} + +/** + * SQL condition "the queue has a working notify trigger": it exists, is enabled and calls the pgmq function. + * One definition for {@see ensureNotifyInsert()} and {@see isNotifyInsertEnabled()}. + * + * @internal + * @param non-empty-string $queueExpression SQL expression with the queue name (a parameter or a PL/pgSQL variable) + */ +function notifyTriggerValidSql(string $queueExpression): string +{ + return \sprintf( + <<<'SQL' + EXISTS ( + SELECT 1 + FROM pg_trigger tg + JOIN pg_class c ON c.oid = tg.tgrelid + JOIN pg_namespace n ON n.oid = c.relnamespace + WHERE n.nspname = 'pgmq' + AND c.relname = pgmq.format_table_name(%s, 'q') + AND tg.tgname = 'trigger_notify_queue_insert_listeners' + AND tg.tgenabled IN ('O', 'A') + AND tg.tgfoid = 'pgmq.notify_queue_listeners()'::regprocedure + ) + SQL, + $queueExpression, ); } diff --git a/tests/PgmqTest.php b/tests/PgmqTest.php index f6130b5..4863003 100644 --- a/tests/PgmqTest.php +++ b/tests/PgmqTest.php @@ -408,6 +408,18 @@ public function testEnsureNotifyInsertRecreatesTriggerCallingAnotherFunction(): self::assertTrue($queue->isNotifyInsertEnabled()); } + public function testEnsureNotifyInsertWorksInsideAnOpenTransaction(): void + { + $queue = createQueue($this->pg, $this->randomQueueName()); + + $tx = $this->pg->beginTransaction(); + ensureNotifyInsert($tx, $queue->name); + self::assertTrue(isNotifyInsertEnabled($tx, $queue->name)); + $tx->rollback(); + + self::assertFalse($queue->isNotifyInsertEnabled(), 'the setup belongs to the caller\'s transaction'); + } + public function testEnsureNotifyInsertChangesThrottleWithoutRecreatingTrigger(): void { $queue = createQueue($this->pg, $this->randomQueueName());