symfony · rtek · Apr 9, 2022 · Apr 9, 2022 · Apr 10, 2022 · Apr 11, 2022
diff --git a/.github/workflows/integration-tests.yml b/.github/workflows/integration-tests.yml
@@ -161,6 +161,9 @@ jobs:
          ./phpunit install
          echo "::endgroup::"

+      - name: Start MySQL
+        run: sudo systemctl start mysql.service
+
      - name: Run tests
        run: ./phpunit --group integration -v
        env:
@@ -174,6 +177,9 @@ jobs:
          MESSENGER_SQS_FIFO_QUEUE_DSN: "sqs://localhost:9494/messages.fifo?sslmode=disable&poll_timeout=0.01"
          KAFKA_BROKER: 127.0.0.1:9092
          POSTGRES_HOST: localhost
+          MYSQL_DSN: mysql:host=localhost # ubuntu-20.04 mysql defaults
+          MYSQL_USERNAME: root
+          MYSQL_PASSWORD: root

      #- name: Run HTTP push tests
      #  if: matrix.php == '8.1'

@@ -1,5 +1,8 @@
 CHANGELOG
 =========
+6.2
+---
+* Add `MysqlStore` based on MySQL `GET_LOCK()` functionality

 6.0
 ---

@@ -223,6 +223,7 @@ private function prune(): void
    private function getCurrentTimestampStatement(): string
    {
        $platform = $this->conn->getDatabasePlatform();
+
        return match (true) {
            $platform instanceof \Doctrine\DBAL\Platforms\MySQLPlatform,
            $platform instanceof \Doctrine\DBAL\Platforms\MySQL57Platform => 'UNIX_TIMESTAMP()',

@@ -0,0 +1,162 @@
+<?php
+
+/*
+ * This file is part of the Symfony package.
+ *
+ * (c) Fabien Potencier <fabien@symfony.com>
+ *
+ * For the full copyright and license information, please view the LICENSE
+ * file that was distributed with this source code.
+ */
+
+namespace Symfony\Component\Lock\Store;
+
+use Symfony\Component\Lock\Exception\InvalidArgumentException;
+use Symfony\Component\Lock\Exception\LockAcquiringException;
+use Symfony\Component\Lock\Exception\LockConflictedException;
+use Symfony\Component\Lock\Key;
+use Symfony\Component\Lock\PersistingStoreInterface;
+
+/**
+ * @author rtek
+ * @author Jérôme TAMARELLE <jerome@tamarelle.net>
+ */
+class MysqlStore implements PersistingStoreInterface
+{
+    private \PDO $conn;
+
+    private string $dsn;
+
+    private array $options;
+
+    private int $connectionId;
+
+    private \PDOStatement $saveStmt;
+
+    private \PDOStatement $existsStmt;
+
+    private \PDOStatement $deleteStmt;
+
+    public function __construct(\PDO|string $connOrDsn, array $options = [])
+    {
+        if ($connOrDsn instanceof \PDO) {
+            $this->conn = $connOrDsn;
+            $this->assertMysqlDriver();
+            if (\PDO::ERRMODE_EXCEPTION !== $this->conn->getAttribute(\PDO::ATTR_ERRMODE)) {
+                throw new InvalidArgumentException(sprintf('"%s" requires PDO error mode attribute be set to throw Exceptions (i.e. $pdo->setAttribute(PDO::ATTR_ERRMODE, PDO::ERRMODE_EXCEPTION)).', __METHOD__));
+            }
+        } else {
+            $this->dsn = $connOrDsn;
+        }
+
+        $this->options = $options;
+    }
+
+    public function save(Key $key): void
+    {
+        if ($this->exists($key)) {
+            return;
+        }
+
+        $stmt = $this->saveStmt ??
+            $this->saveStmt = $this->getConnection()->prepare('SELECT IF(IS_USED_LOCK(:name) = CONNECTION_ID(), -1, GET_LOCK(:name, 0))');
+
+        $name = self::getLockName($key);
+        $stmt->execute(['name' => $name]);
+        $result = $stmt->fetchColumn();
+
+        // lock acquired
+        if (1 === $result) {
+            $key->setState($this->getStateKey($key), $name);
+
+            return;
+        }
+
+        if (0 === $result) {
+            throw new LockConflictedException('Lock already acquired by other connection.');
+        }
+
+        if (-1 === $result) {
+            throw new LockConflictedException('Lock already acquired by this connection.');
+        }
+
+        throw new LockAcquiringException('Failed to acquire lock due to mysql error.');
+    }
+
+    public function putOffExpiration(Key $key, float $ttl): void
+    {
+        // noop - GET_LOCK() does not have a ttl
+    }
+
+    public function delete(Key $key): void
+    {
+        $stmt = $this->deleteStmt ??
+            $this->deleteStmt = $this->getConnection()->prepare('DO RELEASE_LOCK(:name)');
+
+        $stmt->execute(['name' => self::getLockName($key)]);
+
+        $key->removeState($this->getStateKey($key));
+    }
+
+    public function exists(Key $key): bool
+    {
+        $stateKey = $this->getStateKey($key);
+        if (!$key->hasState($stateKey)) {
+            return false;
+        }
+
+        $stmt = $this->existsStmt ??
+            $this->existsStmt = $this->getConnection()->prepare('SELECT IS_USED_LOCK(:name) = CONNECTION_ID()');
+
+        $stmt->execute(['name' => self::getLockName($key)]);
+        $result = $stmt->fetchColumn();
+
+        if (1 !== $result) {
+            $key->removeState($stateKey);
+
+            return false;
+        }
+
+        return true;
+    }
+
+    private function getConnection(): \PDO
+    {
+        if (!isset($this->conn)) {
+            $this->conn = new \PDO(
+                $this->dsn,
+                $this->options['db_username'] ?? null,
+                $this->options['db_password'] ?? null,
+                $this->options['db_connection_options'] ?? null
+            );
+            $this->conn->setAttribute(\PDO::ATTR_ERRMODE, \PDO::ERRMODE_EXCEPTION);
+            $this->assertMysqlDriver();
+        }
+
+        return $this->conn;
+    }
+
+    private function assertMysqlDriver(): void
+    {
+        if ('mysql' !== $driver = $this->conn->getAttribute(\PDO::ATTR_DRIVER_NAME)) {
+            throw new InvalidArgumentException(sprintf('The adapter "%s" does not support the "%s" driver.', __CLASS__, $driver));
+        }
+    }
+
+    private function getStateKey(Key $key): string
+    {
+        if (!isset($this->connectionId)) {
+            $this->connectionId = $this->getConnection()->query('SELECT CONNECTION_ID()')->fetchColumn();
+        }
+
+        return __CLASS__.'_'.$this->connectionId;
+    }
+
+    private static function getLockName(Key $key): string
+    {
+        // mysql limits lock name length to 64 chars
+        $name = (string) $key;
+
+        return \strlen($name) > 64 ? hash('xxh128', $name) : $name;
+    }
+}
@@ -0,0 +1,110 @@
+<?php
+
+namespace Symfony\Component\Lock\Tests\Store;
+
+use Symfony\Component\Lock\Exception\InvalidArgumentException;
+use Symfony\Component\Lock\Exception\LockConflictedException;
+use Symfony\Component\Lock\Key;
+use Symfony\Component\Lock\PersistingStoreInterface;
+use Symfony\Component\Lock\Store\MysqlStore;
+
+class MysqlStoreTest extends AbstractStoreTest
+{
+    protected function getEnv(): array
+    {
+        if (!$dsn = getenv('MYSQL_DSN')) {
+            $this->markTestSkipped('Missing MYSQL_DSN env variable');
+        }
+
+        if (!$user = getenv('MYSQL_USERNAME')) {
+            $this->markTestSkipped('Missing MYSQL_USERNAME env variable');
+        }
+
+        if (!$pass = getenv('MYSQL_PASSWORD')) {
+            $this->markTestSkipped('Missing MYSQL_PASSWORD env variable');
+        }
+
+        return [$dsn, $user, $pass];
+    }
+
+    protected function getPdo(): \PDO
+    {
+        [$dsn, $user, $pass] = $this->getEnv();
+
+        return new \PDO($dsn, $user, $pass);
+    }
+
+    protected function getStore(): PersistingStoreInterface
+    {
+        return new MysqlStore($this->getPdo());
+    }
+
+    public function testDriverRequirement()
+    {
+        $this->expectException(InvalidArgumentException::class);
+        new MysqlStore(new \PDO('sqlite::memory:'));
+    }
+
+    public function testExceptionModeRequirement()
+    {
+        $this->expectException(InvalidArgumentException::class);
+        $pdo = $this->getPdo();
+        $pdo->setAttribute(\PDO::ATTR_ERRMODE, \PDO::ERRMODE_SILENT);
+        new MysqlStore($pdo);
+    }
+
+    public function testOtherConnConflictException()
+    {
+        $storeA = $this->getStore();
+        $storeB = $this->getStore();
+
+        $key = new Key('foo');
+        $storeA->save($key);
+
+        $this->assertFalse($storeB->exists($key));
+
+        try {
+            $storeB->save($key);
+            $this->fail('Expected exception: '.LockConflictedException::class);
+        } catch (LockConflictedException $e) {
+            $this->assertStringContainsString('acquired by other', $e->getMessage());
+        }
+    }
+
+    public function testExistsOnKeyClone()
+    {
+        $store = $this->getStore();
+
+        $key = new Key('foo');
+        $store->save($key);
+
+        $this->assertTrue($store->exists($key));
+        $this->assertTrue($store->exists(clone $key));
+    }
+
+    public function testStoresAreStateless()
+    {
+        $pdo = $this->getPdo();
+
+        $storeA = new MysqlStore($pdo);
+        $storeB = new MysqlStore($pdo);
+        $key = new Key('foo');
+
+        $storeA->save($key);
+        $this->assertTrue($storeA->exists($key));
+        $this->assertTrue($storeB->exists($key));
+
+        $storeB->delete($key);
+        $this->assertFalse($storeB->exists($key));
+        $this->assertFalse($storeA->exists($key));
+    }
+
+    public function testDsnConstructor()
+    {
+        $this->expectNotToPerformAssertions();
+
+        [$host, $user, $pass] = $this->getEnv();
+        $store = new MysqlStore("mysql:$host", ['db_username' => $user, 'db_password' => $pass]);
+        $store->save(new Key('foo'));
+    }
+}