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

[iGrog]

Fixes #20.

Reworked (see the comment below). The first version only skipped the DDL when notifications looked enabled. Concurrent cold starts could still race into it and fail with trigger … already exists; a disabled trigger or one calling another function counted as enabled; and a different throttle interval recreated the trigger.

Problem

Consumer::consume() called enableNotifyInsert() on every start. That drops and recreates the trigger on pgmq.q_<queue>, taking a table lock: a (re)starting consumer waits for every open transaction that inserted into the queue, blocks new inserts meanwhile, and consumers starting together race into the same DDL.

pgmq.notify_insert_throttle is UNLOGGED: crash recovery (or a promoted replica) empties it while the trigger survives, and the trigger then never notifies. Recreating everything on every start used to heal this by accident, so the fix has to keep that case working.

Change

• ensureNotifyInsert(queue, ?throttleInterval) (+ Queue::ensureNotifyInsert()), as one statement (a DO block):

  1. pgmq.acquire_queue_lock(queue): the lock pgmq itself takes to create and drop the queue, so concurrent starts pass one by one;
  2. read the catalog: trigger exists, is enabled (tgenabled O/A) and calls pgmq.notify_queue_listeners(); the throttle row is present;
  3. all in place → nothing; another interval → pgmq.update_notify_insert() (no DDL); otherwise → pgmq.enable_notify_insert().

  Being one statement, it needs no transaction of its own. It works on any PostgresLink: a pool, a connection, or an open transaction (then the queue lock is held until that transaction ends). It costs one round trip.

• isNotifyInsertEnabled(queue, ?throttleInterval) (+ Queue::isNotifyInsertEnabled()): the same check, read-only. The trigger-validity condition is defined once and shared by both.

• Consumer::consume() uses ensureNotifyInsert(). enableNotifyInsert() keeps its "(re)create" semantics.

• The default throttle (30 ms) is shared through an internal notifyThrottleMs() instead of being repeated.

thesis/message-bus needs the same behaviour in its PgmqTransport; thesis-php/message-bus#7 calls this function.

Tests

test	without this change
testConsumerStartDoesNotWaitForProducerTransactions: notifications set up, a producer keeps an insert open, consume() must return within 1 s	❌ CancelledException
testConcurrentColdStartsCreateTheTriggerOnce: 4 consumers (separate pools, like worker processes) start at once on a queue without notifications while a producer forces their starts to overlap; CREATE TRIGGER counted with an event trigger must be exactly 1	❌ trigger … already exists
testEnsureNotifyInsertRestoresLostThrottleRow	✅ the first version handled it too
testEnsureNotifyInsertRecreatesDisabledTrigger	❌
testEnsureNotifyInsertRecreatesTriggerCallingAnotherFunction	❌
testEnsureNotifyInsertChangesThrottleWithoutRecreatingTrigger (0 CREATE TRIGGER)	❌
testEnsureNotifyInsertKeepsExistingSetupAndEnablesMissingOne	✅
testEnsureNotifyInsertWorksInsideAnOpenTransaction (and is rolled back with it)	new behaviour

The event-trigger tests need a superuser connection, which the compose setup provides.

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

scenario	first version	now
8 consumers start at once on a queue without notifications (also with open producer transactions)	failures: already exists	trigger created once, no failures
restart while a producer keeps an insert open	✅	✅
13 rolling restarts under load, 1500 messages	✅	✅ no loss/duplicates/DDL
throttle row lost / trigger dropped	✅	✅
trigger disabled / calls another function	❌	✅
different throttle interval	DDL	✅ no DDL
queue setup concurrent with consumer starts and producers	failures	✅
real crash under load (3 queues × 2 workers restarting at once), two crashes in a row	not measured	✅ notifications restored, one DDL per queue, no loss, no deadlocks

Checks (run locally in the project's Docker images)

PHP 8.3 / 8.4 / 8.5 × lowest / highest: tests ✅ (43), phpstan ✅. fixer-check, rector-check, composer checks ✅. (On PHP 8.5 + lowest deps, the same two amphp/amp deprecations as on 0.1.x; exit code 0.)

Independent of #21. Both add use Amp\TimeoutCancellation; to the test; git merges them cleanly in either order (verified).

[iGrog]

Reworked after testing with a stress harness (details in #20):

• check and change now run in one transaction under pgmq.acquire_queue_lock(), so concurrent cold starts no longer race. Before, they failed with trigger … already exists;
• the trigger counts as valid only when enabled and calling pgmq.notify_queue_listeners();
• a missing throttle row (UNLOGGED, emptied by crash recovery) re-enables notifications;
• a different throttle interval uses update_notify_insert(), without DDL.

New tests: concurrent cold start (event trigger counts CREATE TRIGGER, must be 1), a disabled trigger, a trigger calling another function, a changed interval without DDL. All were checked to fail on the previous version of this PR, except the lost-row test, which it already passed. The whole CI matrix (PHP 8.3–8.5 × lowest/highest) is green locally.

[iGrog]

One more change: ensureNotifyInsert() now runs lock, check and change as one statement (a DO block) instead of opening its own transaction. It works on any PostgresLink, including an already open transaction, where the first version's beginTransaction() is not always allowed, e.g. behind wrappers that own transaction control. It also costs one round trip instead of five. A new test covers the call inside an open transaction. The concurrent-start test fails when the acquire_queue_lock() line is removed (checked), and the whole CI matrix plus the stress harness are green again.