Repository: symfony/symfony
File: src/Symfony/Component/Messenger/Bridge/Doctrine/EventListener/PostgreSqlNotifyOnIdleListener.php (PHP)

Code before:
        }

        foreach ($matched as $connection) {
            // Only the first (active) connection executes LISTEN on the database;
            // the others just mark get() as externally handled to avoid
            // accumulating unread notifications on connections that never
            // call waitForNotify().
            $connection->listen(null === $this->activeConnection);
            $this->activeConnection ??= $connection;
        }
    }

    public function onWorkerRunning(WorkerRunningEvent $event): void
    {
        if (!$event->isWorkerIdle() || !$this->activeConnection) {
            return;
        }

        $config = $this->activeConnection->getConfiguration();

        if (0 >= $timeout = $config['get_notify_timeout']) {
            return;
        }

        // Cap by earliest delayed message: wake up in time to process it
        if (null !== $earliest = $this->activeConnection->getEarliestDelayedMessageTime()) {
            $now = $this->clock->now();
            $deadline = ($earliest->format('U.u') - $now->format('U.u')) * 1000;
            if (0 >= $timeout = min($timeout, $deadline)) {
                return;
            }
        }

        // Cap by worker deadline (--time-limit)
        if (null !== $this->deadline) {
            $deadline = ($this->deadline - $this->clock->now()->format('U.u')) * 1000;
            if (0 >= $timeout = min($timeout, $deadline)) {
                return;
            }
        }

        // Cap by sleep duration when non-PG transports are present to ensure they are still polled regularly
        if (0 >= $timeout = min($timeout, $this->sleepCapMs ?? $timeout)) {
            return;
        }

        $this->logger?->debug('Worker waiting for PostgreSQL LISTEN/NOTIFY wake-up.', ['timeout_ms' => $timeout]);

        $this->activeConnection->waitForNotify((int) $timeout);
    }


Review comment (suggestion):
The `Cap by earliest delayed message` check should "interrogate" _all_ related PG transportNames, not only the "activeConnection"'s. In other words, imagine the following case:

1. There is a delayed queue task in 60 seconds on "queue" (i.e. "transportName") "high". That "queue" is "tracked" by the "activeConnection".
2. There is a delayed queue task in 30 seconds on queue "low".

The result should be: the worker should "pg-sleep" for 30 seconds, not 60 seconds.

There are two potential approaches here:

1. `getEarliestDelayedMessageTime()` is called in a `foreach ($this->connections as $connection) {}` loop. The downside is that there will be n `SELECT MIN(available_at) FROM %s WHERE queue_name = ? AND available_at > ? AND delivered_at IS NULL` queries run every time `WorkerRunningEvent` is emitted. It's acceptable but perhaps it worthwhile to avoid it.
2. `getEarliestDelayedMessageTime()` should not be in `PostgreSqlConnection`. It should be put it in this class, and the SQL rewritten to `SELECT MIN(available_at) FROM %s WHERE queue_name IN (?) AND available_at > ? AND delivered_at IS NULL` (note the `queue_name IN (?)` condition). I would prefer this course of action.

Code after:
        $this->sleepCapMs = \count($matched) < \count($allTransportNames) ? (int) ($event->getIdleTimeout() / 1000) : null;

        if (\count($matched) > 1) {
            $this->validateConnections($matched);
        }

        foreach ($matched as $connection) {
            // Only the first (active) connection executes LISTEN on the database; the others just mark get() as externally
            // handled to avoid accumulating unread notifications on connections that never call waitForNotify().
            $connection->listen(null === $this->activeConnection);
            $this->activeConnection ??= $connection;
        }
    }

    public function onWorkerRunning(WorkerRunningEvent $event): void
    {
        if (!$event->isWorkerIdle() || !$this->activeConnection) {
            return;
        }

        $config = $this->activeConnection->getConfiguration();

        if (0 >= $timeout = $config['get_notify_timeout']) {
            return;
        }

        // Cap by worker deadline (--time-limit) — checked first to avoid a SQL query
        // in getEarliestDelayedMessageTime() when the worker is already past its deadline.
        if (null !== $this->deadline) {
            $deadline = ($this->deadline - $this->clock->now()->format('U.u')) * 1000;
            if (0 >= $timeout = min($timeout, $deadline)) {
                return;
            }
        }

        // Cap by earliest delayed message across all PG queues: wake up in time to process it
        if (null !== $earliest = $this->getEarliestDelayedMessageTime()) {
            $now = $this->clock->now();
            $msUntilEarliest = ($earliest->format('U.u') - $now->format('U.u')) * 1000;
            if (0 >= $timeout = min($timeout, $msUntilEarliest)) {
                return;
            }
        }

        // Cap by sleep duration when non-PG transports are present to ensure they are still polled regularly
        if (0 >= $timeout = (int) min($timeout, $this->sleepCapMs ?? $timeout)) {
            return;
        }

        $this->logger?->debug('Worker waiting for PostgreSQL LISTEN/NOTIFY wake-up.', ['timeout_ms' => $timeout]);