From cf21162232d3b69952119ebb4e03caa7ec8c052c Mon Sep 17 00:00:00 2001
From: max <igrog@kyra.fi>
Date: Wed, 7 Oct 2026 12:38:02 +0300
Subject: [PATCH 1/3] Pgmq: set up insert notifications in createQueue(), not
 on every consumer start

startConsumer() called enable_notify_insert(), 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.

The trigger is now created once with the queue; the consumer only listens.
---
 src/MessageBus/Pgmq/PgmqTransport.php | 8 ++++++--
 1 file changed, 6 insertions(+), 2 deletions(-)

diff --git a/src/MessageBus/Pgmq/PgmqTransport.php b/src/MessageBus/Pgmq/PgmqTransport.php
index 66bdca6..a68b6ff 100644
--- a/src/MessageBus/Pgmq/PgmqTransport.php
+++ b/src/MessageBus/Pgmq/PgmqTransport.php
@@ -58,7 +58,11 @@ public function __construct(
     public function createQueue(string $name): void
     {
         Pgmq\createExtension($this->postgres);
-        Pgmq\createQueue($this->postgres, $name);
+        $queue = Pgmq\createQueue($this->postgres, $name);
+
+        // Notifications are part of the queue setup, not of a consumer start: (re)creating the trigger takes a table
+        // lock that waits for every open transaction inserting into the queue and blocks new inserts meanwhile.
+        $queue->enableNotifyInsert();
 
         $this->postgres->query('create schema if not exists thesis_message_bus;');
         $this->postgres->query(
@@ -189,7 +193,7 @@ public function startConsumer(string $queue, ConsumerHandler $handler): PgmqCons
             $timeoutWatcher,
             new ChannelWatcher(
                 $polls,
-                $this->postgres->listen($queue->enableNotifyInsert()),
+                $this->postgres->listen(Pgmq\channelName($queue->name)),
                 $timeoutWatcher,
             ),
         ]);

From 71deac31605ea91a94ac33e14de42938695cf900 Mon Sep 17 00:00:00 2001
From: max <igrog@kyra.fi>
Date: Wed, 7 Oct 2026 14:03:47 +0300
Subject: [PATCH 2/3] Pgmq: ensure insert notifications in createQueue() and on
 consumer start

Uses thesis/pgmq ensureNotifyInsert(): it reads the catalog when notifications are in
place (no table lock, no DDL), restores them when lost (crash recovery empties the
UNLOGGED throttle table), and serializes concurrent starts under the queue lock.
---
 src/MessageBus/Pgmq/PgmqTransport.php | 9 +++++----
 1 file changed, 5 insertions(+), 4 deletions(-)

diff --git a/src/MessageBus/Pgmq/PgmqTransport.php b/src/MessageBus/Pgmq/PgmqTransport.php
index a68b6ff..f806f3a 100644
--- a/src/MessageBus/Pgmq/PgmqTransport.php
+++ b/src/MessageBus/Pgmq/PgmqTransport.php
@@ -60,9 +60,7 @@ public function createQueue(string $name): void
         Pgmq\createExtension($this->postgres);
         $queue = Pgmq\createQueue($this->postgres, $name);
 
-        // Notifications are part of the queue setup, not of a consumer start: (re)creating the trigger takes a table
-        // lock that waits for every open transaction inserting into the queue and blocks new inserts meanwhile.
-        $queue->enableNotifyInsert();
+        $queue->ensureNotifyInsert();
 
         $this->postgres->query('create schema if not exists thesis_message_bus;');
         $this->postgres->query(
@@ -193,7 +191,10 @@ public function startConsumer(string $queue, ConsumerHandler $handler): PgmqCons
             $timeoutWatcher,
             new ChannelWatcher(
                 $polls,
-                $this->postgres->listen(Pgmq\channelName($queue->name)),
+                // Not enableNotifyInsert(): recreating the trigger on every start took a table lock that waited for open
+                // producer transactions. ensureNotifyInsert() only reads the catalog when notifications are in place,
+                // and restores them when lost (the throttle table is UNLOGGED and is emptied by crash recovery).
+                $this->postgres->listen($queue->ensureNotifyInsert()),
                 $timeoutWatcher,
             ),
         ]);

From 8891a0b0f822ec34b8fc8d1cf0138f23f13d7ebc Mon Sep 17 00:00:00 2001
From: max <igrog@kyra.fi>
Date: Wed, 7 Oct 2026 12:42:21 +0300
Subject: [PATCH 3/3] Remove the @phpstan-ignore that no longer matches

Recent PHPStan 2.2.x infers the nested property assignment correctly, so the ignore
is unmatched and phpstan fails on every cell of the CI matrix (8.4/8.5 x lowest/highest).

(cherry picked from commit ec4557352169cfe1a978542690ccd9bfbedfba04)
---
 src/MessageBus/Handling/Handlers.php | 1 -
 1 file changed, 1 deletion(-)

diff --git a/src/MessageBus/Handling/Handlers.php b/src/MessageBus/Handling/Handlers.php
index b55ba5f..9aacd48 100644
--- a/src/MessageBus/Handling/Handlers.php
+++ b/src/MessageBus/Handling/Handlers.php
@@ -54,7 +54,6 @@ public function with(string $messageClass, callable $handler, string $qualifier
         }
 
         $copy = clone $this;
-        /** @phpstan-ignore assign.propertyType */
         $copy->handlers[$messageClass][$qualifier] = $handler;
 
         return $copy;