Skip to content

[Messenger] Add SKIP LOCKED to the query that retrieves messages #52639

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

Merged
merged 1 commit into from
Nov 21, 2023
Merged

[Messenger] Add SKIP LOCKED to the query that retrieves messages #52639

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
5 changes: 5 additions & 0 deletions src/Symfony/Component/Messenger/Bridge/Doctrine/CHANGELOG.md
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,6 +1,11 @@
CHANGELOG
=========

7.1
---

* Use `SKIP LOCKED` in the doctrine transport for MySQL, PostgreSQL and MSSQL

5.1.0
-----

Expand Down
93 changes: 82 additions & 11 deletions 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 @@ -14,10 +14,15 @@
use Doctrine\DBAL\Connection as DBALConnection;
use Doctrine\DBAL\Exception as DBALException;
use Doctrine\DBAL\Platforms\AbstractPlatform;
use Doctrine\DBAL\Platforms\MariaDb1060Platform;
use Doctrine\DBAL\Platforms\MariaDBPlatform;
use Doctrine\DBAL\Platforms\MySQL57Platform;
use Doctrine\DBAL\Platforms\MySQL80Platform;
use Doctrine\DBAL\Platforms\MySQLPlatform;
use Doctrine\DBAL\Platforms\OraclePlatform;
use Doctrine\DBAL\Platforms\PostgreSQL100Platform;
use Doctrine\DBAL\Platforms\PostgreSQL94Platform;
use Doctrine\DBAL\Platforms\PostgreSQLPlatform;
use Doctrine\DBAL\Platforms\SQLServer2012Platform;
use Doctrine\DBAL\Platforms\SQLServerPlatform;
use Doctrine\DBAL\Query\QueryBuilder;
Expand Down Expand Up @@ -391,28 +396,94 @@ class_exists(MySQLPlatform::class) ? new MySQLPlatform() : new MySQL57Platform()
'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 1 FOR UPDATE',
];

if (class_exists(MySQL80Platform::class) && !method_exists(QueryBuilder::class, 'forUpdate')) {
yield 'MySQL8 & DBAL<3.8' => [
new MySQL80Platform(),
'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 1 FOR UPDATE',
];
}

if (class_exists(MySQL80Platform::class) && method_exists(QueryBuilder::class, 'forUpdate')) {
yield 'MySQL8 & DBAL>=3.8' => [
new MySQL80Platform(),
'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 1 FOR UPDATE SKIP LOCKED',
];
}

yield 'MariaDB' => [
new MariaDBPlatform(),
'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 1 FOR UPDATE',
];

yield 'SQL Server' => [
class_exists(SQLServerPlatform::class) && !class_exists(SQLServer2012Platform::class) ? new SQLServerPlatform() : new SQLServer2012Platform(),
'SELECT m.* FROM messenger_messages m WITH (UPDLOCK, ROWLOCK) WHERE (m.queue_name = ?) AND (m.delivered_at is null OR m.delivered_at < ?) AND (m.available_at <= ?) ORDER BY available_at ASC OFFSET 0 ROWS FETCH NEXT 1 ROWS ONLY ',
];
if (class_exists(MariaDb1060Platform::class)) {
yield 'MariaDB106' => [
new MariaDb1060Platform(),
'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 1 FOR UPDATE SKIP LOCKED',
];
}

if (!class_exists(MySQL57Platform::class)) {
// DBAL >= 4
yield 'Oracle' => [
new OraclePlatform(),
'SELECT w.id AS "id", w.body AS "body", w.headers AS "headers", w.queue_name AS "queue_name", w.created_at AS "created_at", w.available_at AS "available_at", w.delivered_at AS "delivered_at" FROM messenger_messages w WHERE w.id IN (SELECT m.id 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 FETCH NEXT 1 ROWS ONLY) FOR UPDATE',
if (class_exists(MySQL57Platform::class)) {
yield 'Postgres & DBAL<4' => [
new PostgreSQLPlatform(),
'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 1 FOR UPDATE',
];
} else {
// DBAL < 4
yield 'Oracle' => [
yield 'Postgres & DBAL>=4' => [
new PostgreSQLPlatform(),
'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 1 FOR UPDATE SKIP LOCKED',
];
}

if (class_exists(PostgreSQL94Platform::class)) {
yield 'Postgres94' => [
new PostgreSQL94Platform(),
'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 1 FOR UPDATE',
];
}

if (class_exists(PostgreSQL100Platform::class) && !method_exists(QueryBuilder::class, 'forUpdate')) {
yield 'Postgres10 & DBAL<3.8' => [
new PostgreSQL100Platform(),
'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 1 FOR UPDATE',
];
}

if (class_exists(PostgreSQL100Platform::class) && method_exists(QueryBuilder::class, 'forUpdate')) {
yield 'Postgres10 & DBAL>=3.8' => [
new PostgreSQL100Platform(),
'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 1 FOR UPDATE SKIP LOCKED',
];
}

if (!method_exists(QueryBuilder::class, 'forUpdate')) {
yield 'SQL Server & DBAL<3.8' => [
class_exists(SQLServerPlatform::class) && !class_exists(SQLServer2012Platform::class) ? new SQLServerPlatform() : new SQLServer2012Platform(),
'SELECT m.* FROM messenger_messages m WITH (UPDLOCK, ROWLOCK) WHERE (m.queue_name = ?) AND (m.delivered_at is null OR m.delivered_at < ?) AND (m.available_at <= ?) ORDER BY available_at ASC OFFSET 0 ROWS FETCH NEXT 1 ROWS ONLY ',
];
}

if (method_exists(QueryBuilder::class, 'forUpdate')) {
yield 'SQL Server & DBAL>=3.8' => [
class_exists(SQLServerPlatform::class) && !class_exists(SQLServer2012Platform::class) ? new SQLServerPlatform() : new SQLServer2012Platform(),
'SELECT m.* FROM messenger_messages m WITH (UPDLOCK, ROWLOCK, READPAST) WHERE (m.queue_name = ?) AND (m.delivered_at is null OR m.delivered_at < ?) AND (m.available_at <= ?) ORDER BY available_at ASC OFFSET 0 ROWS FETCH NEXT 1 ROWS ONLY ',
];
}

if (!method_exists(QueryBuilder::class, 'forUpdate')) {
yield 'Oracle & DBAL<3.8' => [
new OraclePlatform(),
'SELECT w.id AS "id", w.body AS "body", w.headers AS "headers", w.queue_name AS "queue_name", w.created_at AS "created_at", w.available_at AS "available_at", w.delivered_at AS "delivered_at" FROM messenger_messages w WHERE w.id IN (SELECT a.id FROM (SELECT m.id 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) a WHERE ROWNUM <= 1) FOR UPDATE',
];
} elseif (class_exists(MySQL57Platform::class)) {
Copy link
Contributor

Choose a reason for hiding this comment

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

Oracle?

Copy link
Author

Choose a reason for hiding this comment

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

No, this was already there, it's a way to know if we are using DBAL 4 or not.
Since MySQL57Platform is deprecated, it will not be included in DBAL4 anymore.
I guess we could/should put this in a method with a meaningful name...
I can do that if/when I have time, otherwise, I think it's fine to leave it as is since it was already like this.

yield 'Oracle & 3.8<=DBAL<4' => [
new OraclePlatform(),
'SELECT w.id AS "id", w.body AS "body", w.headers AS "headers", w.queue_name AS "queue_name", w.created_at AS "created_at", w.available_at AS "available_at", w.delivered_at AS "delivered_at" FROM messenger_messages w WHERE w.id IN (SELECT a.id FROM (SELECT m.id 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) a WHERE ROWNUM <= 1) FOR UPDATE SKIP LOCKED',
];
} else {
yield 'Oracle & DBAL>=4' => [
new OraclePlatform(),
'SELECT w.id AS "id", w.body AS "body", w.headers AS "headers", w.queue_name AS "queue_name", w.created_at AS "created_at", w.available_at AS "available_at", w.delivered_at AS "delivered_at" FROM messenger_messages w WHERE w.id IN (SELECT m.id 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 FETCH NEXT 1 ROWS ONLY) FOR UPDATE SKIP LOCKED',
];
}
}

Expand Down
55 changes: 36 additions & 19 deletions 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 @@ -18,6 +18,7 @@
use Doctrine\DBAL\LockMode;
use Doctrine\DBAL\Platforms\AbstractMySQLPlatform;
use Doctrine\DBAL\Platforms\OraclePlatform;
use Doctrine\DBAL\Query\ForUpdate\ConflictResolutionMode;
use Doctrine\DBAL\Query\QueryBuilder;
use Doctrine\DBAL\Result;
use Doctrine\DBAL\Schema\Schema;
Expand All @@ -32,6 +33,8 @@
*
* @author Vincent Touzet <vincent.touzet@gmail.com>
* @author Kévin Dunglas <dunglas@gmail.com>
* @author Herberto Graca <herberto.graca@gmail.com>
* @author Alexander Malyk <shu.rick.ifmo@gmail.com>
*/
class Connection implements ResetInterface
{
Expand Down Expand Up @@ -178,28 +181,22 @@ public function get(): ?array
->where('w.id IN ('.str_replace('SELECT a.* FROM', 'SELECT a.id FROM', $sql).')')
->setParameters($query->getParameters());

if (method_exists(QueryBuilder::class, 'forUpdate')) {
$query->forUpdate();
}

$sql = $query->getSQL();
} elseif (method_exists(QueryBuilder::class, 'forUpdate')) {
$query->forUpdate();
try {
$sql = $query->getSQL();
} catch (DBALException $e) {
}
} elseif (preg_match('/FROM (.+) WHERE/', (string) $sql, $matches)) {
$fromClause = $matches[1];
$sql = str_replace(
sprintf('FROM %s WHERE', $fromClause),
sprintf('FROM %s WHERE', $this->driverConnection->getDatabasePlatform()->appendLockHint($fromClause, LockMode::PESSIMISTIC_WRITE)),
$sql
);
}

// use SELECT ... FOR UPDATE to lock table
if (!method_exists(QueryBuilder::class, 'forUpdate')) {
if (method_exists(QueryBuilder::class, 'forUpdate')) {
$sql = $this->addLockMode($query, $sql);
} else {
if (preg_match('/FROM (.+) WHERE/', (string) $sql, $matches)) {
$fromClause = $matches[1];
$sql = str_replace(
sprintf('FROM %s WHERE', $fromClause),
sprintf('FROM %s WHERE', $this->driverConnection->getDatabasePlatform()->appendLockHint($fromClause, LockMode::PESSIMISTIC_WRITE)),
$sql
);
}

// use SELECT ... FOR UPDATE to lock table
$sql .= ' '.$this->driverConnection->getDatabasePlatform()->getWriteLockSQL();
}

Expand Down Expand Up @@ -493,4 +490,24 @@ private function updateSchema(): void
}
}
}

private function addLockMode(QueryBuilder $query, string $sql): string
{
$query->forUpdate(ConflictResolutionMode::SKIP_LOCKED);
try {
return $query->getSQL();
} catch (DBALException) {
return $this->fallBackToForUpdate($query, $sql);
}
}

private function fallBackToForUpdate(QueryBuilder $query, string $sql): string
{
$query->forUpdate();
try {
return $query->getSQL();
} catch (DBALException) {
return $sql;
}
}
}