[Messenger][Doctrine] cleanup only the configured queue_name

[smatyas]

Symfony version(s) affected

4.4-7.1

Description

Using a master-master mysql replica, a couple of transports with different queue_name-s and separate consumers by transports, it can easily happen that the cleanup statements will result in replication error due to preventable race-conditions.

If we run distinct consumers against the replicated dbs, it should not happen, if the cleanup queries would filter for the queue name. Currently every consumer deletes old records even from unrelated queues.

I know that we could workaround this problem with separate tables, but I think it might be a nice and correct addition for the current implementation.

The current implementation does not filter for the 'queu_name', but it should:

-                 $this->driverConnection->delete($this->configuration['table_name'], ['delivered_at' => '9999-12-31 23:59:59']);
+                $this->driverConnection->delete($this->configuration['table_name'], ['delivered_at' => '9999-12-31 23:59:59', 'queue_name' => $this->configuration['queue_name']]);

How to reproduce

1. create a master-master mysql replica
2. create multiple transport using the same table

# messenger.yaml
# ...
        transports:
            job_a:
                dsn: '%env(MESSENGER_TRANSPORT_DSN)%?queue_name=job_a'
            job_b:
                dsn: '%env(MESSENGER_TRANSPORT_DSN)%?queue_name=job_b'

3. consume job_a from db-server-a using multiple consumers
4. consume job_b from db-server-b using multiple consumers

Possible Solution

Do the cleanup only for the configured queue_name.

In Symfony\Component\Messenger\Bridge\Doctrine\Transport::get

-                 $this->driverConnection->delete($this->configuration['table_name'], ['delivered_at' => '9999-12-31 23:59:59']);
+                $this->driverConnection->delete($this->configuration['table_name'], ['delivered_at' => '9999-12-31 23:59:59', 'queue_name' => $this->configuration['queue_name']]);

Additional Context

We experienced this issue with symfony 4.4, but I can see from the code that it still works the same way.