Consumer::consume() recreates the notify trigger on every start: table lock, blocked producers, deadlocks #20

Updated after further testing — see the comment below. The first version of this issue (and of #22) missed two cases: concurrent starts on a queue without notifications still race, and notifications are silently lost after crash recovery.

Summary

With listenForInserts: true (the default), every Consumer::consume() call runs enableNotifyInsert(). pgmq.enable_notify_insert() drops and recreates the trigger on the queue table (DROP TRIGGER …; CREATE CONSTRAINT TRIGGER …). This is DDL that takes a table-level lock on pgmq.q_<queue>:

• a (re)starting consumer waits for every open transaction that has inserted into the queue to finish;
• while it waits, new inserts into that queue queue up behind it;
• consumers of the same queue starting at the same time race into the same DDL. They wait for each other or deadlock, and the loser fails with trigger "trigger_notify_queue_insert_listeners" … already exists.

Consumers restart routinely (deploys, worker recycling after N messages or a memory limit), so this is not limited to the first start.

A second problem: the throttle row is UNLOGGED

The insert trigger notifies only when it finds the queue's row in pgmq.notify_insert_throttle, and that table is UNLOGGED. After crash recovery, and on a promoted replica, the table is empty while the trigger survives. From then on the trigger silently never notifies, and consumers fall back to polling.

Today this heals by accident, because every consumer start recreates everything. Any fix that stops recreating the trigger on start has to keep this case working.

Where

• src/Consumer.php L51: $channelName = $queue->enableNotifyInsert();
• pgmq.enable_notify_insert() (pgmq extension): pgmq.disable_notify_insert() (DROP TRIGGER IF EXISTS …, delete the throttle row), then insert the row and CREATE CONSTRAINT TRIGGER …, every time.

Reproduction

$queue = createQueue($pg, 'q');
$queue->enableNotifyInsert();                       // already set up

$producer = $pg->beginTransaction();                // a producer inside its transaction
send($producer, 'q', new SendMessage('{}'));

async(static fn() => createConsumer($pg)->consume(static function (): void {}, new ConsumeConfig('q')))
    ->await(new TimeoutCancellation(1));            // CancelledException: blocked until the producer commits

Lost notifications: DELETE FROM pgmq.notify_insert_throttle WHERE queue_name = 'q' (what crash recovery does to the whole table), then insert into the queue: no notification arrives until the throttle row is recreated.

Proposed fix (#22)

ensureNotifyInsert(queue, ?throttleInterval) makes notifications work while touching the trigger only when needed. One transaction:

1. pgmq.acquire_queue_lock(queue): the same lock pgmq takes to create and drop the queue. Consumers starting together pass one by one instead of racing into the DDL;
2. read the state from the catalog: the trigger exists, is enabled (tgenabled O/A) and calls pgmq.notify_queue_listeners(); the throttle row is present;
3. decide:
   • all in place with the same interval → nothing (catalog reads only, no lock on the queue table);
   • in place with another interval → pgmq.update_notify_insert() (no DDL);
   • otherwise (missing, disabled, foreign trigger, or the throttle row is gone) → pgmq.enable_notify_insert().

Consumer::consume() uses it. isNotifyInsertEnabled() exposes the same check. enableNotifyInsert() keeps its "(re)create" semantics.

How it was verified

A stress harness on real Postgres, against pgmq 1.11.1 (the image from this repo's compose) and 1.13.0. Crashes are real: kill -9 of a backend makes the postmaster run crash recovery. Workers and producers are separate processes.