[Messenger] PostgreSQL: new messages are not picked up for 60 seconds

[bendavies]

Symfony version(s) affected

5.3 onwards

Description

Hi,

There appears to be a bug in PostgreSqlConnection meaning that newly available messages are not picked immediately.

This is due to this line:

symfony/src/Symfony/Component/Messenger/Bridge/Doctrine/Transport/PostgreSqlConnection.php

Line 90 in 6cdfd71

(microtime(true) * 1000 - $this->queueEmptiedAt < $this->configuration['check_delayed_interval'])

I really don't understand what that line is doing apart from being something to do with delayed message, but apart from that don't understand it.

How to reproduce

1. start your consume: bin/console messenger:consume async -vvv
2. queue some job via any means you like
3. this message will be consumed immediately
4. queue another message via any means you like
5. this message and all other new ones will not be picked up for 60 seconds after the message in 2. was consumed.

Simply: after the queue is emptied, new messages will not be picked up for 60 seconds because of check_delayed_interval' => 60000

Possible Solution

Unsure, no idea what

symfony/src/Symfony/Component/Messenger/Bridge/Doctrine/Transport/PostgreSqlConnection.php

Line 90 in 6cdfd71

(microtime(true) * 1000 - $this->queueEmptiedAt < $this->configuration['check_delayed_interval'])

is meant to be doing.

Additional Context

Found a related discussion #36990

[bendavies]

Secondly, and this is possibly related, but it seems to be that these default settings simply do not work and make NOTIFY support not work:

'use_notify' => true,
'get_notify_timeout' => 0

since get_notify_timeout is zero, this line

symfony/src/Symfony/Component/Messenger/Bridge/Doctrine/Transport/PostgreSqlConnection.php

Line 85 in 6cdfd71

$notification = $wrappedConnection->pgsqlGetNotify(\PDO::FETCH_ASSOC, $this->configuration['get_notify_timeout']);

returns immedately and never "waits" for a notification, which seems to defeat the point of using listen/notify completely.

edited:
problem discovered here: #45056 (comment)

[Sajito]

I had the issue too, that the consumer picked up new messages only after 60 seconds. I also thought it might be due to the get_notify_timeout option, but it turned out having this option set to 0 works perfectly fine.
After some time I checked how the NOTIFY/LISTEN feature actually works and found that there should be a trigger on the messenger_messages table, which was missing for me. Have you checked if the trigger exists for you?

The 60 seconds delay is the default polling rate, when NOTIFY/LISTEN feature is not used. In my case the problem is actually sitting in this check, maybe it's the same in your case:

symfony/src/Symfony/Component/Messenger/Bridge/Doctrine/Transport/DoctrineTransportFactory.php

Line 51 in acf1f71

if ($useNotify && $driverConnection->getDriver() instanceof AbstractPostgreSQLDriver) {

I have installed the sentry/sentry-symfony bundle, which decorated the connection's driver, therefore the check $driverConnection->getDriver() instanceof AbstractPostgreSQLDriver fails and the NOTIFY feature is never actually set up.

[bendavies]

yes we have the trigger, and our problem is not sentry.

as per the OP, this line is the problem:

symfony/src/Symfony/Component/Messenger/Bridge/Doctrine/Transport/PostgreSqlConnection.php

Line 90 in 6cdfd71

(microtime(true) * 1000 - $this->queueEmptiedAt < $this->configuration['check_delayed_interval'])

once the queue is emptied, new messages won't be checked for 60 seconds.

[Sajito]

That's not what that line does, because you're ignoring the first condition. The check is only delayed, if there is no notification or the notification is not related to the checked queue. Check the whole if and notice the && at the end of line 88.

symfony/src/Symfony/Component/Messenger/Bridge/Doctrine/Transport/PostgreSqlConnection.php

Lines 86 to 91 in 6cdfd71

	if (
	// no notifications, or for another table or queue
	(false === $notification || $notification['message'] !== $this->configuration['table_name'] || $notification['payload'] !== $this->configuration['queue_name']) &&
	// delayed messages
	(microtime(true) * 1000 - $this->queueEmptiedAt < $this->configuration['check_delayed_interval'])
	) {

If a notification for the queue is received, then the first part will fail and the delay-condition won't be checked.

So your issue must be caused either by the notification not being triggered, or the listen not being notified.

Edit: The method itself by default is actually called every second, which also if configurable with the sleep option.

[bendavies]

@Sajito thanks for your reply.

I've investigated again, and what I am finding is that the listener is being "lost".

the listener is setup here:

symfony/src/Symfony/Component/Messenger/Bridge/Doctrine/Transport/PostgreSqlConnection.php

Line 72 in 6cdfd71

$this->executeStatement(sprintf('LISTEN "%s"', $this->configuration['table_name']));

just above the pgsqlGetNotify line i have added:

$listeners = $this->driverConnection->fetchAssociative('select * from pg_listening_channels();');

which correctly shows the listener the first time around.
on the 2nd time around, the listener is gone, and pg_listening_channels() returns false.

this appears to be why pgsqlGetNotify isnt' working.

now, need to figure out what's happening to the listener.

[bendavies]

and.. it's the doctrine_close_connection (DoctrineCloseConnectionMiddleware) middleware.

so the PostgreSqlConnection needs to detect that it is no longer listening and relisten

[dunglas]

Thanks for investigating @bendavies! Do you want to work on a patch?

[bendavies]

@dunglas yes, but...choices:

1. PostgreSqlConnection checks every single call to get if it is listening or not using pg_listening_channels()
2. PostgreSqlConnection listens on Doctrine\DBAL\Events::postConnect and relistens if required. there is unfortunately no event for closing the connection.
3. another way?

any preferences?

[goetas]

I have the same issue. What is the harm of querying LISTEN xxxx every the time in get? anyway is per-session and calling LISTEN multiple times is idempotent.

https://www.postgresql.org/docs/current/sql-listen.html
If the current session is already registered as a listener for this notification channel, nothing is done.

In that way it would be possible to stop keeping track of "are we listening or not" and remove the $this->listening prop. (eventually we can also UNLISTEN as soon as pgsqlGetNotify completes).

[bendavies]

Nice find. That sounds good

[bendavies]

I'll work on this next week

[mtomala]

@bendavies Hi. Any updates on this?

[goetas]

#47209 is the pull request implementing #45056 (comment)