PgmqTransport: every consumer start recreates the notify trigger (table lock, blocked producers, deadlocks) #3

Updated after further testing — see the comment below. The first proposal (set notifications up only in createQueue()) loses them for good after crash recovery; the current proposal ensures them under the queue lock.

Summary

PgmqTransport::startConsumer() calls $queue->enableNotifyInsert(), and pgmq.enable_notify_insert() drops and recreates the trigger on pgmq.q_<queue>. That DDL takes a table-level lock:

• a (re)starting worker waits for every open transaction that has inserted into the queue, and in the meantime blocks new inserts into it;
• workers of the same queue starting together race into the same DDL and can deadlock.

Workers restart often: deploys, and recycling after N messages or a memory limit, which happens every few seconds under load. Each restart repeats the DDL.

Where

src/MessageBus/Pgmq/PgmqTransport.php L192: $this->postgres->listen($queue->enableNotifyInsert()),

Observed

Two workers of one queue were recycled in the same tick after reaching their message budget:

Consumer 'queue_a#1' exited (code=0) after 253s
Consumer 'queue_a#2' exited (code=0) after 253s
Starting consumer: queue_a#1
Starting consumer: queue_a#2
Consumer 'queue_a#1' crashed: Amp\Postgres\PostgresQueryError: ERROR:  deadlock detected

What a fix must keep: notifications lost after crash recovery

The insert trigger notifies only when it finds the queue's row in pgmq.notify_insert_throttle, and that table is UNLOGGED. Crash recovery, or a promoted replica, leaves it empty while the trigger survives, and notifications stop silently. The current code heals this by accident: every worker start recreates everything.

We saw it ourselves: after a crash restart of Postgres, all queues of an application quietly switched to 5-second polling.

Reproduction

$transport = new PgmqTransport($postgres);
$transport->createQueue('q');

$producer = $postgres->beginTransaction();           // a producer inside its transaction
$transport->dispatchInTransaction($producer, [$envelopeToQ]);

async(static fn() => $transport->startConsumer('q', $handler))
    ->await(new TimeoutCancellation(1));              // CancelledException: waits for the producer's commit

Proposed fix

• createQueue() and startConsumer() both call thesis/pgmq's ensureNotifyInsert() (Do not recreate the notify trigger on consumer start; ensure notifications under the queue lock pgmq#22). In one transaction under pgmq.acquire_queue_lock() it reads the catalog and changes something only when needed:
  • nothing, when the trigger is valid and the throttle row is in place (the usual restart: no DDL, no lock on the queue table);
  • update_notify_insert() for another interval;
  • enable_notify_insert() when anything is missing, disabled or foreign. That covers crash recovery and concurrent cold starts, which pass one by one.
• The pgmq-specific knowledge (trigger, throttle table, locks) stays in thesis/pgmq; the transport only calls it.

This needs a thesis/pgmq release with thesis-php/pgmq#22. PR: #7 (draft until then). Integration tests: #8.

Verified with a stress harness (real Postgres, pgmq 1.11.1 and 1.13.0, real crashes via kill -9)