diff --git a/src/ConsumeConfig.php b/src/ConsumeConfig.php index f02604b..e7f08cd 100644 --- a/src/ConsumeConfig.php +++ b/src/ConsumeConfig.php @@ -19,7 +19,7 @@ /** * @param non-empty-string $queue * @param positive-int $batch - * @param bool $listenForInserts will call {@see enableNotifyInsert} and will listen for notifications from the channel, which significantly optimizes the number of requests to the Postgres server for new messages. It is recommended to enable it. + * @param bool $listenForInserts will call {@see ensureNotifyInsert} and will listen for notifications from the channel, which significantly optimizes the number of requests to the Postgres server for new messages. It is recommended to enable it. */ public function __construct( public string $queue, diff --git a/src/Consumer.php b/src/Consumer.php index a6bd5de..99f00b4 100644 --- a/src/Consumer.php +++ b/src/Consumer.php @@ -48,7 +48,7 @@ public function consume( $watchers[] = $timeoutWatcher; if ($config->listenForInserts) { - $channelName = $queue->enableNotifyInsert(); + $channelName = $queue->ensureNotifyInsert(); $watchers[] = new Internal\ChannelWatcher( $polls, diff --git a/src/Queue.php b/src/Queue.php index fba37af..0307f68 100644 --- a/src/Queue.php +++ b/src/Queue.php @@ -295,6 +295,27 @@ public function enableNotifyInsert(?TimeSpan $throttleInterval = null): string ); } + /** + * @return non-empty-string + */ + public function ensureNotifyInsert(?TimeSpan $throttleInterval = null): string + { + return ensureNotifyInsert( + pg: $this->pg, + queue: $this->name, + throttleInterval: $throttleInterval, + ); + } + + public function isNotifyInsertEnabled(?TimeSpan $throttleInterval = null): bool + { + return isNotifyInsertEnabled( + pg: $this->pg, + queue: $this->name, + throttleInterval: $throttleInterval, + ); + } + public function disableNotifyInsert(): void { disableNotifyInsert( diff --git a/src/queue.fn.php b/src/queue.fn.php index 545e79f..ceacd03 100644 --- a/src/queue.fn.php +++ b/src/queue.fn.php @@ -461,12 +461,136 @@ function enableNotifyInsert( ): string { $pg->execute('SELECT pgmq.enable_notify_insert(:queue_name, :throttle_interval_ms)', [ 'queue_name' => $queue, - 'throttle_interval_ms' => ($throttleInterval ?? TimeSpan::fromMilliseconds(30))->toMilliseconds(), + 'throttle_interval_ms' => notifyThrottleMs($throttleInterval), ]); return channelName($queue); } +/** + * Makes insert notifications work for the queue, touching the trigger only when it is really needed: + * + * * already enabled with this throttle interval → nothing (catalog reads only, no lock on the queue table); + * * enabled with another interval → `pgmq.update_notify_insert()` (no DDL); + * * missing, disabled or calling another function, or the throttle row is gone → `pgmq.enable_notify_insert()`. + * + * The throttle row lives in an UNLOGGED table: crash recovery (or a promoted replica) empties it while the trigger + * survives, and the trigger then never notifies. That is why the row is checked, not only the trigger. + * + * Check and change run as ONE statement under `pgmq.acquire_queue_lock()` (the lock pgmq itself takes to create and + * drop the queue), so consumers starting together do not race into the DDL: the first one sets notifications up, the + * others wait for it and then find them in place. Being one statement, it needs no transaction of its own and works + * on any link: a pool, a connection or an open transaction (then the lock is held until that transaction ends). + * + * @api + * @param non-empty-string $queue + * @return non-empty-string + */ +function ensureNotifyInsert( + PostgresLink $pg, + string $queue, + ?TimeSpan $throttleInterval = null, +): string { + $pg->query(\sprintf( + <<<'SQL' + DO $ensure$ + DECLARE + v_queue text := %s; + v_throttle_ms integer := %d; + v_current_ms integer; + BEGIN + PERFORM pgmq.acquire_queue_lock(v_queue); + + SELECT t.throttle_interval_ms INTO v_current_ms + FROM pgmq.notify_insert_throttle t + WHERE t.queue_name = v_queue; + + IF NOT (%s) OR v_current_ms IS NULL THEN + PERFORM pgmq.enable_notify_insert(v_queue, v_throttle_ms); + ELSIF v_current_ms <> v_throttle_ms THEN + PERFORM pgmq.update_notify_insert(v_queue, v_throttle_ms); + END IF; + END + $ensure$ + SQL, + $pg->quoteLiteral($queue), + notifyThrottleMs($throttleInterval), + notifyTriggerValidSql('v_queue'), + )); + + return channelName($queue); +} + +/** + * Whether insert notifications work for the queue with this throttle interval: the trigger exists, is enabled and + * calls `pgmq.notify_queue_listeners()`, and the throttle row is in place. Reads the catalog only. + * + * @api + * @param non-empty-string $queue + */ +function isNotifyInsertEnabled( + PostgresLink $pg, + string $queue, + ?TimeSpan $throttleInterval = null, +): bool { + /** @var array{enabled: bool} $row */ + $row = $pg + ->execute( + \sprintf( + <<<'SQL' + SELECT (%s) AND EXISTS ( + SELECT 1 + FROM pgmq.notify_insert_throttle t + WHERE t.queue_name = :queue_name AND t.throttle_interval_ms = :throttle_interval_ms + ) AS enabled + SQL, + notifyTriggerValidSql(':queue_name'), + ), + [ + 'queue_name' => $queue, + 'throttle_interval_ms' => notifyThrottleMs($throttleInterval), + ], + ) + ->fetchRow(); + + return $row['enabled']; +} + +/** + * SQL condition "the queue has a working notify trigger": it exists, is enabled and calls the pgmq function. + * One definition for {@see ensureNotifyInsert()} and {@see isNotifyInsertEnabled()}. + * + * @internal + * @param non-empty-string $queueExpression SQL expression with the queue name (a parameter or a PL/pgSQL variable) + */ +function notifyTriggerValidSql(string $queueExpression): string +{ + return \sprintf( + <<<'SQL' + EXISTS ( + SELECT 1 + FROM pg_trigger tg + JOIN pg_class c ON c.oid = tg.tgrelid + JOIN pg_namespace n ON n.oid = c.relnamespace + WHERE n.nspname = 'pgmq' + AND c.relname = pgmq.format_table_name(%s, 'q') + AND tg.tgname = 'trigger_notify_queue_insert_listeners' + AND tg.tgenabled IN ('O', 'A') + AND tg.tgfoid = 'pgmq.notify_queue_listeners()'::regprocedure + ) + SQL, + $queueExpression, + ); +} + +/** + * @internal + */ +function notifyThrottleMs(?TimeSpan $throttleInterval): int +{ + return ($throttleInterval ?? TimeSpan::fromMilliseconds(30))->toMilliseconds(); +} + /** * @api * @param non-empty-string $queue diff --git a/tests/PgmqTest.php b/tests/PgmqTest.php index ce2f1f0..4863003 100644 --- a/tests/PgmqTest.php +++ b/tests/PgmqTest.php @@ -8,10 +8,12 @@ use Amp\Postgres\PostgresConnection; use Amp\Postgres\PostgresConnectionPool; use Amp\Postgres\PostgresQueryError; +use Amp\TimeoutCancellation; use PHPUnit\Framework\Attributes\CoversClass; use PHPUnit\Framework\Attributes\DoesNotPerformAssertions; use PHPUnit\Framework\TestCase; use Thesis\Time\TimeSpan; +use function Amp\async; use function Amp\delay; #[CoversClass(Queue::class)] @@ -276,6 +278,159 @@ public function testEnableDisableNotifies(): void $queue->disableNotifyInsert(); } + public function testConsumerStartDoesNotWaitForProducerTransactions(): void + { + $queue = createQueue($this->pg, $this->randomQueueName()); + $queue->enableNotifyInsert(); + + // A producer is in the middle of a transaction that has already inserted into the queue. + $producer = $this->pg->beginTransaction(); + send($producer, $queue->name, new SendMessage(self::TESTING_MESSAGE)); + + $pg = $this->pg; + + try { + // A worker (re)starts: notifications are already set up, so it must not touch the trigger. Recreating it + // takes a table lock that waits for every open producer transaction and blocks new inserts meanwhile. + /** @var ConsumeContext $context */ + $context = async(static fn(): ConsumeContext => createConsumer($pg)->consume( + static function (): void {}, + new ConsumeConfig($queue->name), + ))->await(new TimeoutCancellation(1)); + + $context->stop(); + $context->awaitCompletion(); + } finally { + $producer->rollback(); + } + + self::assertTrue($queue->isNotifyInsertEnabled()); + } + + public function testEnsureNotifyInsertKeepsExistingSetupAndEnablesMissingOne(): void + { + $queue = createQueue($this->pg, $this->randomQueueName()); + + self::assertFalse($queue->isNotifyInsertEnabled()); + self::assertSame(channelName($queue->name), $queue->ensureNotifyInsert()); + self::assertTrue($queue->isNotifyInsertEnabled()); + + // Already enabled with the same throttle: nothing to do. + self::assertSame(channelName($queue->name), $queue->ensureNotifyInsert()); + self::assertTrue($queue->isNotifyInsertEnabled()); + + // A different throttle is a different setup: it is applied. + $queue->ensureNotifyInsert(TimeSpan::fromSeconds(1)); + self::assertTrue($queue->isNotifyInsertEnabled(TimeSpan::fromSeconds(1))); + self::assertFalse($queue->isNotifyInsertEnabled()); + + $queue->disableNotifyInsert(); + self::assertFalse($queue->isNotifyInsertEnabled(TimeSpan::fromSeconds(1))); + } + + public function testConcurrentColdStartsCreateTheTriggerOnce(): void + { + $queue = createQueue($this->pg, $this->randomQueueName()); // notifications not set up yet + + // A producer keeps an insert into the queue open, so that every consumer start has to wait for the table lock + // and the starts are guaranteed to overlap. + $producer = $this->pg->beginTransaction(); + send($producer, $queue->name, new SendMessage(self::TESTING_MESSAGE)); + + $dsn = (string) getenv('THESIS_PGMQ_DSN'); + $creations = $this->countTriggerCreations(static function () use ($dsn, $queue, $producer): void { + $starts = []; + + for ($i = 0; $i < 4; ++$i) { + // Each consumer has its own pool, as separate worker processes do (one pool cannot LISTEN twice). + $pool = new PostgresConnectionPool(PostgresConfig::fromString($dsn)); + $starts[] = async(static fn(): ConsumeContext => createConsumer($pool)->consume( + static function (): void {}, + new ConsumeConfig($queue->name), + )); + } + + delay(0.5); + $producer->rollback(); + + foreach ($starts as $start) { + /** @var ConsumeContext $context */ + $context = $start->await(new TimeoutCancellation(5)); + $context->stop(); + $context->awaitCompletion(); + } + }, $queue->name); + + self::assertSame(1, $creations, 'concurrent starts must not race into enable_notify_insert()'); + self::assertTrue($queue->isNotifyInsertEnabled()); + } + + public function testEnsureNotifyInsertRestoresLostThrottleRow(): void + { + $queue = createQueue($this->pg, $this->randomQueueName()); + $queue->enableNotifyInsert(); + + // pgmq.notify_insert_throttle is UNLOGGED: crash recovery empties it while the trigger survives. + $this->pg->execute('DELETE FROM pgmq.notify_insert_throttle WHERE queue_name = :queue_name', ['queue_name' => $queue->name]); + self::assertFalse($queue->isNotifyInsertEnabled()); + + $queue->ensureNotifyInsert(); + + self::assertTrue($queue->isNotifyInsertEnabled()); + } + + public function testEnsureNotifyInsertRecreatesDisabledTrigger(): void + { + $queue = createQueue($this->pg, $this->randomQueueName()); + $queue->enableNotifyInsert(); + $this->pg->query(\sprintf('ALTER TABLE pgmq.q_%s DISABLE TRIGGER trigger_notify_queue_insert_listeners', $queue->name)); + self::assertFalse($queue->isNotifyInsertEnabled()); + + $queue->ensureNotifyInsert(); + + self::assertTrue($queue->isNotifyInsertEnabled()); + } + + public function testEnsureNotifyInsertRecreatesTriggerCallingAnotherFunction(): void + { + $queue = createQueue($this->pg, $this->randomQueueName()); + $queue->enableNotifyInsert(); + $this->pg->query('CREATE OR REPLACE FUNCTION public.thesis_test_noop_trigger() RETURNS trigger LANGUAGE plpgsql AS $$ BEGIN RETURN NEW; END $$'); + $this->pg->query(\sprintf('DROP TRIGGER trigger_notify_queue_insert_listeners ON pgmq.q_%s', $queue->name)); + $this->pg->query(\sprintf( + 'CREATE CONSTRAINT TRIGGER trigger_notify_queue_insert_listeners AFTER INSERT ON pgmq.q_%s DEFERRABLE FOR EACH ROW EXECUTE PROCEDURE public.thesis_test_noop_trigger()', + $queue->name, + )); + self::assertFalse($queue->isNotifyInsertEnabled()); + + $queue->ensureNotifyInsert(); + + self::assertTrue($queue->isNotifyInsertEnabled()); + } + + public function testEnsureNotifyInsertWorksInsideAnOpenTransaction(): void + { + $queue = createQueue($this->pg, $this->randomQueueName()); + + $tx = $this->pg->beginTransaction(); + ensureNotifyInsert($tx, $queue->name); + self::assertTrue(isNotifyInsertEnabled($tx, $queue->name)); + $tx->rollback(); + + self::assertFalse($queue->isNotifyInsertEnabled(), 'the setup belongs to the caller\'s transaction'); + } + + public function testEnsureNotifyInsertChangesThrottleWithoutRecreatingTrigger(): void + { + $queue = createQueue($this->pg, $this->randomQueueName()); + $queue->enableNotifyInsert(TimeSpan::fromSeconds(1)); + + $creations = $this->countTriggerCreations(static fn() => $queue->ensureNotifyInsert(), $queue->name); + + self::assertSame(0, $creations); + self::assertTrue($queue->isNotifyInsertEnabled()); + } + public function testInvalidConsumerConfiguration(): void { $queue = createQueue($this->pg, $this->randomQueueName()); @@ -637,6 +792,42 @@ public function testValidateTopicPattern(): void validateTopicPattern($this->pg, 'logs.**'); } + /** + * Runs $action and returns how many times the notify trigger of the queue was created meanwhile (event trigger). + * + * @param callable(): mixed $action + * @param non-empty-string $queue + */ + private function countTriggerCreations(callable $action, string $queue): int + { + $this->pg->query('CREATE TABLE IF NOT EXISTS public.thesis_test_ddl (object_identity text)'); + $this->pg->query('TRUNCATE public.thesis_test_ddl'); + $this->pg->query(<<<'SQL' + CREATE OR REPLACE FUNCTION public.thesis_test_log_ddl() RETURNS event_trigger LANGUAGE plpgsql AS $$ + DECLARE r record; + BEGIN + FOR r IN SELECT * FROM pg_event_trigger_ddl_commands() LOOP + INSERT INTO public.thesis_test_ddl (object_identity) VALUES (r.object_identity); + END LOOP; + END $$ + SQL); + $this->pg->query('DROP EVENT TRIGGER IF EXISTS thesis_test_ddl'); + $this->pg->query("CREATE EVENT TRIGGER thesis_test_ddl ON ddl_command_end WHEN TAG IN ('CREATE TRIGGER') EXECUTE FUNCTION public.thesis_test_log_ddl()"); + + try { + $action(); + } finally { + $this->pg->query('DROP EVENT TRIGGER IF EXISTS thesis_test_ddl'); + } + + /** @var array{n: int} $row */ + $row = $this->pg + ->execute('SELECT count(*)::int AS n FROM public.thesis_test_ddl WHERE object_identity LIKE :pattern', ['pattern' => '% on pgmq.q_' . $queue]) + ->fetchRow(); + + return $row['n']; + } + /** * @return non-empty-string */