Merge release 3.19.2 into 3.20.x

docker-compose.yaml

@@ -13,4 +13,12 @@ services:
       - MYSQL_ALLOW_EMPTY_PASSWORD="yes"
       - MYSQL_DATABASE=eventstore
     ports:
-      - 3306:3306
+      - 3306:3306
+
+  mariadb:
+    image: mariadb:12
+    environment:
+      - MYSQL_ALLOW_EMPTY_PASSWORD="yes"
+      - MYSQL_DATABASE=eventstore
+    ports:
+      - 3307:3306

src/Store/DoctrineDbalStore.php

@@ -64,6 +64,14 @@ final class DoctrineDbalStore implements Store, SubscriptionStore, DoctrineSchem
      */
     private const DEFAULT_LOCK_ID = 133742;
 
+    /**
+     * MariaDB does not support an infinite (negative) lock timeout. Very large values such as
+     * PHP_INT_MAX overflow its internal timeout arithmetic and make GET_LOCK return NULL. We
+     * therefore use a large but safe value (INT32_MAX minus a small buffer) as "effectively
+     * infinite" wait.
+     */
+    private const INFINITE_MARIADB_LOCK_TIMEOUT = 2_147_482_647;
+
     private readonly HeadersSerializer $headersSerializer;
 
     /** @var array{table_name: string, aggregate_id_type: 'string'|'uuid', locking: bool, lock_id: int, lock_timeout: int} */
@@ -493,14 +501,28 @@ private function lock(): void
         }
 
         if ($platform instanceof MariaDBPlatform || $platform instanceof MySQLPlatform) {
-            $this->connection->fetchAllAssociative(
+            $lockTimeout = $this->config['lock_timeout'];
+
+            if ($platform instanceof MariaDBPlatform && $lockTimeout < 0) {
+                $lockTimeout = self::INFINITE_MARIADB_LOCK_TIMEOUT;
+            }
+
+            $result = $this->connection->fetchOne(
                 sprintf(
                     'SELECT GET_LOCK("%s", %d)',
                     $this->config['lock_id'],
-                    $this->config['lock_timeout'],
+                    $lockTimeout,
                 ),
             );
 
+            if ($result === 0) {
+                throw LockCouldNotBeAcquired::byTimeout($this->config['lock_id'], $this->config['lock_timeout']);
+            }
+
+            if ($result !== 1) {
+                throw LockCouldNotBeAcquired::byError($this->config['lock_id']);
+            }
+
             return;
         }
 
@@ -522,13 +544,21 @@ private function unlock(): void
         }
 
         if ($platform instanceof MariaDBPlatform || $platform instanceof MySQLPlatform) {
-            $this->connection->fetchAllAssociative(
+            $result = $this->connection->fetchOne(
                 sprintf(
                     'SELECT RELEASE_LOCK("%s")',
                     $this->config['lock_id'],
                 ),
             );
 
+            if ($result === 0) {
+                throw LockCouldNotBeFreed::notOurs($this->config['lock_id']);
+            }
+
+            if ($result !== 1) {
+                throw LockCouldNotBeFreed::notExist($this->config['lock_id']);
+            }
+
             return;
         }
 

src/Store/LockCouldNotBeAcquired.php

@@ -0,0 +1,24 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Patchlevel\EventSourcing\Store;
+
+use function sprintf;
+
+final class LockCouldNotBeAcquired extends StoreException
+{
+    public static function byTimeout(int $id, int $timeout): self
+    {
+        return new self(sprintf(
+            'The lock with id [%s] could not be acquired with a timeout of %d',
+            $id,
+            $timeout,
+        ));
+    }
+
+    public static function byError(int $id): self
+    {
+        return new self(sprintf('There was an error when tried to get the lock with id [%s]', $id));
+    }
+}

src/Store/LockCouldNotBeFreed.php

@@ -0,0 +1,20 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Patchlevel\EventSourcing\Store;
+
+use function sprintf;
+
+final class LockCouldNotBeFreed extends StoreException
+{
+    public static function notExist(int $id): self
+    {
+        return new self(sprintf('The lock with id [%s] could not be freed as it does not exist', $id));
+    }
+
+    public static function notOurs(int $id): self
+    {
+        return new self(sprintf('The lock with id [%s] could not be freed as it is not ours', $id));
+    }
+}

src/Store/StreamDoctrineDbalStore.php

@@ -73,6 +73,14 @@ final class StreamDoctrineDbalStore implements StreamStore, SubscriptionStore, D
      */
     private const DEFAULT_LOCK_ID = 133742;
 
+    /**
+     * MariaDB does not support an infinite (negative) lock timeout. Very large values such as
+     * PHP_INT_MAX overflow its internal timeout arithmetic and make GET_LOCK return NULL. We
+     * therefore use a large but safe value (INT32_MAX minus a small buffer) as "effectively
+     * infinite" wait.
+     */
+    private const INFINITE_MARIADB_LOCK_TIMEOUT = 2_147_482_647;
+
     private readonly HeadersSerializer $headersSerializer;
 
     private readonly ClockInterface $clock;
@@ -572,14 +580,28 @@ private function lock(): void
         }
 
         if ($platform instanceof MariaDBPlatform || $platform instanceof MySQLPlatform) {
-            $this->connection->fetchAllAssociative(
+            $lockTimeout = $this->config['lock_timeout'];
+
+            if ($platform instanceof MariaDBPlatform && $lockTimeout < 0) {
+                $lockTimeout = self::INFINITE_MARIADB_LOCK_TIMEOUT;
+            }
+
+            $result = $this->connection->fetchOne(
                 sprintf(
                     'SELECT GET_LOCK("%s", %d)',
                     $this->config['lock_id'],
-                    $this->config['lock_timeout'],
+                    $lockTimeout,
                 ),
             );
 
+            if ($result === 0) {
+                throw LockCouldNotBeAcquired::byTimeout($this->config['lock_id'], $this->config['lock_timeout']);
+            }
+
+            if ($result !== 1) {
+                throw LockCouldNotBeAcquired::byError($this->config['lock_id']);
+            }
+
             return;
         }
 
@@ -601,13 +623,21 @@ private function unlock(): void
         }
 
         if ($platform instanceof MariaDBPlatform || $platform instanceof MySQLPlatform) {
-            $this->connection->fetchAllAssociative(
+            $result = $this->connection->fetchOne(
                 sprintf(
                     'SELECT RELEASE_LOCK("%s")',
                     $this->config['lock_id'],
                 ),
             );
 
+            if ($result === 0) {
+                throw LockCouldNotBeFreed::notOurs($this->config['lock_id']);
+            }
+
+            if ($result !== 1) {
+                throw LockCouldNotBeFreed::notExist($this->config['lock_id']);
+            }
+
             return;
         }
 

tests/Integration/Store/DoctrineDbalStoreTest.php

@@ -6,12 +6,19 @@
 
 use DateTimeImmutable;
 use Doctrine\DBAL\Connection;
+use Doctrine\DBAL\DriverManager;
+use Doctrine\DBAL\Platforms\PostgreSQLPlatform;
+use Doctrine\DBAL\Platforms\SQLitePlatform;
 use Doctrine\DBAL\Schema\Schema;
 use Patchlevel\EventSourcing\Aggregate\AggregateHeader;
 use Patchlevel\EventSourcing\Message\Message;
 use Patchlevel\EventSourcing\Schema\DoctrineSchemaDirector;
 use Patchlevel\EventSourcing\Serializer\DefaultEventSerializer;
 use Patchlevel\EventSourcing\Store\DoctrineDbalStore;
+use Patchlevel\EventSourcing\Store\Header\PlayheadHeader;
+use Patchlevel\EventSourcing\Store\Header\RecordedOnHeader;
+use Patchlevel\EventSourcing\Store\Header\StreamNameHeader;
+use Patchlevel\EventSourcing\Store\LockCouldNotBeAcquired;
 use Patchlevel\EventSourcing\Store\Store;
 use Patchlevel\EventSourcing\Store\UniqueConstraintViolation;
 use Patchlevel\EventSourcing\Tests\DbalManager;
@@ -20,6 +27,7 @@
 use PHPUnit\Framework\TestCase;
 
 use function json_decode;
+use function sprintf;
 
 #[CoversNothing]
 final class DoctrineDbalStoreTest extends TestCase
@@ -34,6 +42,7 @@ public function setUp(): void
         $this->store = new DoctrineDbalStore(
             $this->connection,
             DefaultEventSerializer::createFromPaths([__DIR__ . '/Events']),
+            config: ['lock_timeout' => 1],
         );
 
         $schemaDirector = new DoctrineSchemaDirector(
@@ -190,6 +199,45 @@ public function testSave10000Messages(): void
         self::assertEquals(10000, $result);
     }
 
+    public function testSaveLockTimeout(): void
+    {
+        if ($this->connection->getDatabasePlatform() instanceof SQLitePlatform) {
+            $this->markTestSkipped('SQLite does not support locks');
+        }
+
+        if ($this->connection->getDatabasePlatform() instanceof PostgreSQLPlatform) {
+            $this->markTestSkipped('PostgreSQL does lock indefinitely');
+        }
+
+        $profileId = ProfileId::generate();
+
+        $messages = [
+            Message::create(new ProfileCreated($profileId, 'test'))
+                ->withHeader(new StreamNameHeader(sprintf('profile-%s', $profileId->toString())))
+                ->withHeader(new PlayheadHeader(1))
+                ->withHeader(new RecordedOnHeader(new DateTimeImmutable('2020-01-01 00:00:00'))),
+        ];
+
+        $connection = DriverManager::getConnection($this->connection->getParams());
+
+        $lock = $connection->fetchOne(
+            sprintf(
+                'SELECT GET_LOCK("%s", %d)',
+                133742,
+                1,
+            ),
+        );
+        self::assertSame(1, $lock);
+
+        $this->expectException(LockCouldNotBeAcquired::class);
+        $this->expectExceptionMessage('The lock with id [133742] could not be acquired with a timeout of 1');
+        try {
+            $this->store->save(...$messages);
+        } finally {
+            $connection->close();
+        }
+    }
+
     public function testLoad(): void
     {
         $profileId = ProfileId::generate();

tests/Integration/Store/StreamDoctrineDbalStoreTest.php

@@ -6,6 +6,9 @@
 
 use DateTimeImmutable;
 use Doctrine\DBAL\Connection;
+use Doctrine\DBAL\DriverManager;
+use Doctrine\DBAL\Platforms\PostgreSQLPlatform;
+use Doctrine\DBAL\Platforms\SQLitePlatform;
 use Doctrine\DBAL\Schema\Schema;
 use Patchlevel\EventSourcing\Clock\FrozenClock;
 use Patchlevel\EventSourcing\Message\Message;
@@ -19,6 +22,7 @@
 use Patchlevel\EventSourcing\Store\Header\PlayheadHeader;
 use Patchlevel\EventSourcing\Store\Header\RecordedOnHeader;
 use Patchlevel\EventSourcing\Store\Header\StreamNameHeader;
+use Patchlevel\EventSourcing\Store\LockCouldNotBeAcquired;
 use Patchlevel\EventSourcing\Store\StreamDoctrineDbalStore;
 use Patchlevel\EventSourcing\Store\StreamStore;
 use Patchlevel\EventSourcing\Store\UniqueConstraintViolation;
@@ -51,6 +55,7 @@ public function setUp(): void
             $this->connection,
             DefaultEventSerializer::createFromPaths([__DIR__ . '/Events']),
             clock: $this->clock,
+            config: ['lock_timeout' => 1],
         );
 
         $schemaDirector = new DoctrineSchemaDirector(
@@ -396,6 +401,45 @@ public function testSave10000Messages(): void
         self::assertEquals(10000, $result);
     }
 
+    public function testSaveLockTimeout(): void
+    {
+        if ($this->connection->getDatabasePlatform() instanceof SQLitePlatform) {
+            $this->markTestSkipped('SQLite does not support locks');
+        }
+
+        if ($this->connection->getDatabasePlatform() instanceof PostgreSQLPlatform) {
+            $this->markTestSkipped('PostgreSQL does lock indefinitely');
+        }
+
+        $profileId = ProfileId::generate();
+
+        $messages = [
+            Message::create(new ProfileCreated($profileId, 'test'))
+                ->withHeader(new StreamNameHeader(sprintf('profile-%s', $profileId->toString())))
+                ->withHeader(new PlayheadHeader(1))
+                ->withHeader(new RecordedOnHeader(new DateTimeImmutable('2020-01-01 00:00:00'))),
+        ];
+
+        $connection = DriverManager::getConnection($this->connection->getParams());
+
+        $lock = $connection->fetchOne(
+            sprintf(
+                'SELECT GET_LOCK("%s", %d)',
+                133742,
+                1,
+            ),
+        );
+        self::assertSame(1, $lock);
+
+        $this->expectException(LockCouldNotBeAcquired::class);
+        $this->expectExceptionMessage('The lock with id [133742] could not be acquired with a timeout of 1');
+        try {
+            $this->store->save(...$messages);
+        } finally {
+            $connection->close();
+        }
+    }
+
     public function testLoad(): void
     {
         $profileId = ProfileId::generate();