Do not recreate the notify trigger on consumer start; ensure notifications under the queue lock

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,

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,

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(

src/queue.fn.php

@@ -461,12 +461,136 @@ 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);
 }
 
+/**
+ * Makes insert notifications work for the queue, touching the trigger only when it is really needed:
+ *
+ * * already enabled with this throttle interval → nothing (catalog reads only, no lock on the queue table);
+ * * enabled with another interval → `pgmq.update_notify_insert()` (no DDL);
+ * * missing, disabled or calling another function, or the throttle row is gone → `pgmq.enable_notify_insert()`.
+ *
+ * The throttle row lives in an UNLOGGED table: crash recovery (or a promoted replica) empties it while the trigger
+ * survives, and the trigger then never notifies. That is why the row is checked, not only the trigger.
+ *
+ * Check and change run as ONE statement under `pgmq.acquire_queue_lock()` (the lock pgmq itself takes to create and
+ * drop the queue), so consumers starting together do not race into the DDL: the first one sets notifications up, the
+ * others wait for it and then find them in place. Being one statement, it needs no transaction of its own and works
+ * on any link: a pool, a connection or an open transaction (then the lock is held until that transaction ends).
+ *
+ * @api
+ * @param non-empty-string $queue
+ * @return non-empty-string
+ */
+function ensureNotifyInsert(
+    PostgresLink $pg,
+    string $queue,
+    ?TimeSpan $throttleInterval = null,
+): string {
+    $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);
+}
+
+/**
+ * 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
+ */
+function isNotifyInsertEnabled(
+    PostgresLink $pg,
+    string $queue,
+    ?TimeSpan $throttleInterval = null,
+): bool {
+    /** @var array{enabled: bool} $row */
+    $row = $pg
+        ->execute(
+            \sprintf(
+                <<<'SQL'
+                    SELECT (%s) AND EXISTS (
+                        SELECT 1
+                        FROM pgmq.notify_insert_throttle t
+                        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 $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,
+    );
+}
+
+/**
+ * @internal
+ */
+function notifyThrottleMs(?TimeSpan $throttleInterval): int
+{
+    return ($throttleInterval ?? TimeSpan::fromMilliseconds(30))->toMilliseconds();
+}
+
 /**
  * @api
  * @param non-empty-string $queue

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,159 @@ 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 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 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());
+        $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());
@@ -637,6 +792,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
      */