Pgmq: ensure insert notifications in createQueue() and on consumer start (via thesis/pgmq)

[iGrog]

Fixes #3. Draft: needs a thesis/pgmq release with thesis-php/pgmq#22 (it calls Queue::ensureNotifyInsert()). Until then phpstan reports Call to an undefined method Thesis\Pgmq\Queue::ensureNotifyInsert(); once released, the thesis/pgmq constraint in composer.json should be raised to that version.

Reworked (see the comment below). The first version set notifications up only in createQueue(). Measured with real crashes: that loses them for good after crash recovery, because the throttle table is UNLOGGED.

Problem

startConsumer() called enableNotifyInsert(), which drops and recreates the trigger on the queue table on every start. That takes a table lock: a (re)starting worker waits for every open transaction that inserted into the queue, blocks new inserts meanwhile, and workers starting together race into the DDL. Observed: deadlock detected after two workers were recycled in the same tick.

Change

• createQueue() and startConsumer() call $queue->ensureNotifyInsert() from thesis/pgmq. In one transaction under pgmq.acquire_queue_lock() it:
  • does nothing when the trigger is valid and the throttle row is in place (the usual restart: catalog reads only);
  • calls update_notify_insert() for another interval;
  • calls enable_notify_insert() when anything is missing, disabled or foreign: crash recovery, a queue created before, concurrent cold starts (they pass one by one).
• Knowledge about pgmq internals (trigger, throttle table, locks) stays in thesis/pgmq; the transport only calls it.
• It also carries the one-line phpstan fix from Remove the @phpstan-ignore that no longer matches #5 (identical change, merges in either order).

Verified

Integration tests in #8 (red on 0.5.x, green with #6 + this PR + thesis-php/pgmq#22, verified locally against that pgmq branch):

• consumerStartDoesNotWaitForProducerTransactions;
• concurrentConsumerStartsCreateTheTriggerOnce (event trigger counts CREATE TRIGGER; 4 workers with separate pools);
• consumerStartRestoresNotificationsLostInCrashRecovery;
• consumerStartRestoresDisabledTrigger;
• createdQueueWakesConsumerOnInsert.

Stress harness (real Postgres, pgmq 1.11.1 and 1.13.0, real crashes via kill -9):

scenario	0.5.x	first version	now
restart while a producer keeps an insert open	❌ blocked 1.02 s	✅	✅
8 workers start at once on a queue without notifications	❌ 4 DDL, already exists, deadlock detected	❌ never set up	✅ one DDL
13 rolling restarts under load	❌ 11 DDL, 5 deadlocks	✅	✅ no DDL
throttle row lost / trigger disabled or foreign	✅	❌	✅
real crash under load, 3 queues × 2 workers	✅ 5 DDL per queue	❌ notifications lost	✅ one DDL per queue, no loss, no deadlocks

[iGrog]

Reworked and turned into a draft. The first version (notifications only in createQueue()) loses them for good after crash recovery, as #3 now explains. This version calls thesis/pgmq's ensureNotifyInsert() (thesis-php/pgmq#22) from createQueue() and startConsumer(), so the pgmq-specific logic lives in one place, in thesis/pgmq.

It needs a thesis/pgmq release with #22, hence the draft: until then phpstan reports the missing method. Verified locally against that pgmq branch: the tests in #8 pass 14/14, and the stress harness is green on pgmq 1.11.1 and 1.13.0, including real crashes.

[iGrog]

CI status, for clarity: tests are green. phpstan reports Call to an undefined method Thesis\Pgmq\Queue::ensureNotifyInsert() only because the released thesis/pgmq (0.1.7) does not contain thesis-php/pgmq#22 yet. The listen() expects non-empty-string, mixed given error follows from that. This PR turns green once a thesis/pgmq release with #22 exists and the constraint in composer.json is raised.

Suggested merge order: thesis-php/pgmq#21, thesis-php/pgmq#22 → thesis/pgmq release → #5, #6 → #7 (bump thesis/pgmq) → #8 (its 3 failing tests turn green with #6 and #7).