Skip to content

Navigation Menu

Sign in
Appearance settings

Search code, repositories, users, issues, pull requests...

Provide feedback

We read every piece of feedback, and take your input very seriously.

Saved searches

Use saved searches to filter your results more quickly
Appearance settings

[Doctrine][Messenger] Remove old MySQL special handling that causes deadlocks #61963

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
wants to merge 1 commit into
base: 7.4
Choose a base branch
Loading
from
+17 −33
Open

[Doctrine][Messenger] Remove old MySQL special handling that causes deadlocks #61963

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
16 changes: 16 additions & 0 deletions 16 src/Symfony/Component/Messenger/Bridge/Doctrine/Tests/Transport/ConnectionTest.php
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -779,6 +779,22 @@ public function testConfigureSchema()
$connection = new Connection(['table_name' => 'queue_table'], $driverConnection);
$connection->configureSchema($schema, $driverConnection, fn () => true);
$this->assertTrue($schema->hasTable('queue_table'));

// Ensure the covering index for the SELECT query exists
$table = $schema->getTable('queue_table');
$hasCoveringIndex = false;
foreach ($table->getIndexes() as $index) {
// Doctrine DBAL 4+: use getIndexedColumns(); fallback to getColumns() for older versions
$columns = method_exists($index, 'getIndexedColumns')
? array_map(static fn ($ic) => $ic->getColumnName()->toString(), $index->getIndexedColumns())
: $index->getColumns();

if ($columns === ['queue_name', 'available_at', 'delivered_at', 'id']) {
$hasCoveringIndex = true;
break;
}
}
$this->assertTrue($hasCoveringIndex, 'Expected covering index on [queue_name, available_at, delivered_at, id] not found');
}

public function testConfigureSchemaDifferentDbalConnection()
Expand Down
34 changes: 1 addition & 33 deletions 34 src/Symfony/Component/Messenger/Bridge/Doctrine/Transport/Connection.php
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -12,11 +12,9 @@
namespace Symfony\Component\Messenger\Bridge\Doctrine\Transport;

use Doctrine\DBAL\Connection as DBALConnection;
use Doctrine\DBAL\Driver\Exception as DriverException;
use Doctrine\DBAL\Exception as DBALException;
use Doctrine\DBAL\Exception\TableNotFoundException;
use Doctrine\DBAL\LockMode;
use Doctrine\DBAL\Platforms\AbstractMySQLPlatform;
use Doctrine\DBAL\Platforms\OraclePlatform;
use Doctrine\DBAL\Platforms\PostgreSQLPlatform;
use Doctrine\DBAL\Query\ForUpdate\ConflictResolutionMode;
Expand Down Expand Up @@ -59,7 +57,6 @@ class Connection implements ResetInterface
protected ?float $queueEmptiedAt = null;

private bool $autoSetup;
private bool $doMysqlCleanup = false;

/**
* Constructor.
Expand All @@ -83,7 +80,6 @@ public function __construct(
public function reset(): void
{
$this->queueEmptiedAt = null;
$this->doMysqlCleanup = false;
}

public function getConfiguration(): array
Expand Down Expand Up @@ -161,19 +157,6 @@ public function send(string $body, array $headers, int $delay = 0): string

public function get(): ?array
{
if ($this->doMysqlCleanup && $this->driverConnection->getDatabasePlatform() instanceof AbstractMySQLPlatform) {
try {
$this->driverConnection->delete($this->configuration['table_name'], ['delivered_at' => '9999-12-31 23:59:59']);
$this->doMysqlCleanup = false;
} catch (DriverException $e) {
// Ignore the exception
} catch (TableNotFoundException $e) {
if ($this->autoSetup) {
$this->setup();
}
}
}

get:
$this->driverConnection->beginTransaction();
try {
Expand Down Expand Up @@ -261,14 +244,6 @@ public function get(): ?array
public function ack(string $id): bool
{
try {
if ($this->driverConnection->getDatabasePlatform() instanceof AbstractMySQLPlatform) {
if ($updated = $this->driverConnection->update($this->configuration['table_name'], ['delivered_at' => '9999-12-31 23:59:59'], ['id' => $id]) > 0) {
$this->doMysqlCleanup = true;
}

return $updated;
}

return $this->driverConnection->delete($this->configuration['table_name'], ['id' => $id]) > 0;
} catch (DBALException $exception) {
throw new TransportException($exception->getMessage(), 0, $exception);
Expand All @@ -278,14 +253,6 @@ public function ack(string $id): bool
public function reject(string $id): bool
{
try {
if ($this->driverConnection->getDatabasePlatform() instanceof AbstractMySQLPlatform) {
if ($updated = $this->driverConnection->update($this->configuration['table_name'], ['delivered_at' => '9999-12-31 23:59:59'], ['id' => $id]) > 0) {
$this->doMysqlCleanup = true;
}

return $updated;
}

return $this->driverConnection->delete($this->configuration['table_name'], ['id' => $id]) > 0;
} catch (DBALException $exception) {
throw new TransportException($exception->getMessage(), 0, $exception);
Expand Down Expand Up @@ -569,6 +536,7 @@ private function addTableToSchema(Schema $schema): void
$table->addIndex(['queue_name']);
Copy link

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Suggested change
$table->addIndex(['queue_name']);

This index is now useless as it's covered by the new index. I suggest to remove it.

Copy link
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Agree. Also other 2 indexes are not needed too, since compound index covers it all

Copy link

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

For precision, the new index doesn't covers the other 2.
Index on columns a,b,c will be used on queries which have where condition on (a), on (a+b) or on (a + b + c). So queries with where b = ? or where c = ? or where b = ? and c = ? won't use the index on (a,b,c).

Now, I still believe we can remove them since there is no queries with condition only on available_at or delivered_at or on (available_at + delivered_at), cf the other comment.#61963 (comment)

$table->addIndex(['available_at']);
$table->addIndex(['delivered_at']);
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Are those other indexes still necessary ? AFAICT, queries never put conditions on only one of those fields. They put conditions on either the primary key only or on the 3 fields together (which will use the new index covering them all).

Copy link
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more. 

+---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+------------+-----------+----------+-------------+
| DIGEST_TEXT                                                                                                                                                                                                                                 | COUNT_STAR | total_sec | lock_sec | avg_lock_ms |
+---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+------------+-----------+----------+-------------+
| SELECT `m` . * FROM `messenger_messages` `m` WHERE ( `m` . `queue_name` = ? ) AND ( `m` . `delivered_at` IS NULL OR `m` . `delivered_at` < ? ) AND ( `m` . `available_at` <= ? ) ORDER BY `available_at` ASC LIMIT ? FOR UPDATE SKIP LOCKED |   73271148 | 55182.812 |   91.843 |       0.001 |
| DELETE FROM `messenger_messages` WHERE `id` = ?                                                                                                                                                                                             |   49961358 | 52783.023 |   97.966 |       0.002 |
| INSERT INTO `messenger_messages` ( `body` , `headers` , `queue_name` , `created_at` , `available_at` ) VALUES (...)                                                                                                                         |   50093799 |  8557.638 |   74.552 |       0.001 |
| UPDATE `messenger_messages` SET `delivered_at` = ? WHERE `id` = ?                                                                                                                                                                           |   50173785 |   6835.58 |   29.569 |       0.001 |

At least for MySQL/MariaDB, yeah. Those indexes are now useless and are not used.

$table->addIndex(['queue_name', 'available_at', 'delivered_at', 'id']);
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@psihius are you sure the ID field is needed?

The select query doesn't use the ID field in the where condition.

Copy link
Author

@psihius psihius Oct 6, 2025 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The query does not, but the primary key is used within the index to lock the column, this makes locking more efficient.
Also btree indexes, at least in MySQL/MariaDB, always contain a primary key part as that points to the row, so I just put it there explictly.
Might not needed for MSSQL/Oracle/PosgtreSQL, but I've decided for the moment not to do any platform dependent code, as it might not be needed. Lets see what the maintainers want to do here and adjust accordingly.


// We need to create a sequence for Oracle and set the id column to get the correct nextval
if ($this->driverConnection->getDatabasePlatform() instanceof OraclePlatform) {
Expand Down
Loading
Morty Proxy This is a proxified and sanitized view of the page, visit original site.