add write lock for event store

DavidBadura · DavidBadura · commit 74c38a0e261b · 2024-07-13T10:14:13.000+02:00
diff --git a/phpstan.neon.dist b/phpstan.neon.dist
@@ -5,4 +5,6 @@ parameters:
 	level: max
 	paths:
 		- src
-	checkGenericClassInNonGenericObjectType: false
+	ignoreErrors:
+		-
+			identifier: missingType.generics
diff --git a/src/Store/DoctrineDbalStore.php b/src/Store/DoctrineDbalStore.php
@@ -7,7 +7,10 @@
 use Closure;
 use Doctrine\DBAL\Connection;
 use Doctrine\DBAL\Exception\UniqueConstraintViolationException;
+use Doctrine\DBAL\Platforms\MariaDBPlatform;
+use Doctrine\DBAL\Platforms\MySQLPlatform;
 use Doctrine\DBAL\Platforms\PostgreSQLPlatform;
+use Doctrine\DBAL\Platforms\SQLitePlatform;
 use Doctrine\DBAL\Query\QueryBuilder;
 use Doctrine\DBAL\Schema\Schema;
 use Doctrine\DBAL\Types\Type;
@@ -48,12 +51,17 @@ final class DoctrineDbalStore implements Store, SubscriptionStore, DoctrineSchem
      */
     private const MAX_UNSIGNED_SMALL_INT = 65_535;
 
+    /**
+     * Default lock id for advisory lock.
+     */
+    private const DEFAULT_LOCK_ID = 133742;
+
     private readonly HeadersSerializer $headersSerializer;
 
-    /** @var array{table_name: string, aggregate_id_type: 'string'|'uuid'} */
+    /** @var array{table_name: string, aggregate_id_type: 'string'|'uuid', locking: bool, lock_id: int, lock_timeout: int} */
     private readonly array $config;
 
-    /** @param array{table_name?: string, aggregate_id_type?: 'string'|'uuid'} $config */
+    /** @param array{table_name?: string, aggregate_id_type?: 'string'|'uuid', locking?: bool, lock_id?: int, lock_timeout?: int} $config */
     public function __construct(
         private readonly Connection $connection,
         private readonly EventSerializer $eventSerializer,
@@ -65,6 +73,9 @@ public function __construct(
         $this->config = array_merge([
             'table_name' => 'eventstore',
             'aggregate_id_type' => 'uuid',
+            'locking' => true,
+            'lock_id' => self::DEFAULT_LOCK_ID,
+            'lock_timeout' => -1,
         ], $config);
     }
 
@@ -155,115 +166,120 @@ public function save(Message ...$messages): void
             return;
         }
 
-        $this->connection->transactional(
-            function (Connection $connection) use ($messages): void {
-                /** @var array<string, int> $achievedUntilPlayhead */
-                $achievedUntilPlayhead = [];
-
-                $booleanType = Type::getType(Types::BOOLEAN);
-                $dateTimeType = Type::getType(Types::DATETIMETZ_IMMUTABLE);
-
-                $columns = [
-                    'aggregate',
-                    'aggregate_id',
-                    'playhead',
-                    'event',
-                    'payload',
-                    'recorded_on',
-                    'new_stream_start',
-                    'archived',
-                    'custom_headers',
-                ];
-
-                $columnsLength = count($columns);
-                $batchSize = (int)floor(self::MAX_UNSIGNED_SMALL_INT / $columnsLength);
-                $placeholder = implode(', ', array_fill(0, $columnsLength, '?'));
-
-                $parameters = [];
-                $placeholders = [];
-                /** @var array<int<0, max>, Type> $types */
-                $types = [];
-                $position = 0;
-                foreach ($messages as $message) {
-                    /** @var int<0, max> $offset */
-                    $offset = $position * $columnsLength;
-                    $placeholders[] = $placeholder;
-
-                    $data = $this->eventSerializer->serialize($message->event());
-
-                    try {
-                        $aggregateHeader = $message->header(AggregateHeader::class);
-                    } catch (HeaderNotFound $e) {
-                        throw new MissingDataForStorage($e->name, $e);
-                    }
+        $this->lock();
+        try {
+            $this->connection->transactional(
+                function (Connection $connection) use ($messages): void {
+                    /** @var array<string, int> $achievedUntilPlayhead */
+                    $achievedUntilPlayhead = [];
+
+                    $booleanType = Type::getType(Types::BOOLEAN);
+                    $dateTimeType = Type::getType(Types::DATETIMETZ_IMMUTABLE);
+
+                    $columns = [
+                        'aggregate',
+                        'aggregate_id',
+                        'playhead',
+                        'event',
+                        'payload',
+                        'recorded_on',
+                        'new_stream_start',
+                        'archived',
+                        'custom_headers',
+                    ];
+
+                    $columnsLength = count($columns);
+                    $batchSize = (int)floor(self::MAX_UNSIGNED_SMALL_INT / $columnsLength);
+                    $placeholder = implode(', ', array_fill(0, $columnsLength, '?'));
 
-                    $parameters[] = $aggregateHeader->aggregateName;
-                    $parameters[] = $aggregateHeader->aggregateId;
-                    $parameters[] = $aggregateHeader->playhead;
-                    $parameters[] = $data->name;
-                    $parameters[] = $data->payload;
+                    $parameters = [];
+                    $placeholders = [];
+                    /** @var array<int<0, max>, Type> $types */
+                    $types = [];
+                    $position = 0;
+                    foreach ($messages as $message) {
+                        /** @var int<0, max> $offset */
+                        $offset = $position * $columnsLength;
+                        $placeholders[] = $placeholder;
 
-                    $parameters[] = $aggregateHeader->recordedOn;
-                    $types[$offset + 5] = $dateTimeType;
+                        $data = $this->eventSerializer->serialize($message->event());
 
-                    $streamStart = $message->hasHeader(StreamStartHeader::class);
+                        try {
+                            $aggregateHeader = $message->header(AggregateHeader::class);
+                        } catch (HeaderNotFound $e) {
+                            throw new MissingDataForStorage($e->name, $e);
+                        }
 
-                    if ($streamStart) {
-                        $key = $aggregateHeader->aggregateName . '/' . $aggregateHeader->aggregateId;
-                        $achievedUntilPlayhead[$key] = $aggregateHeader->playhead;
-                    }
+                        $parameters[] = $aggregateHeader->aggregateName;
+                        $parameters[] = $aggregateHeader->aggregateId;
+                        $parameters[] = $aggregateHeader->playhead;
+                        $parameters[] = $data->name;
+                        $parameters[] = $data->payload;
 
-                    $parameters[] = $streamStart;
-                    $types[$offset + 6] = $booleanType;
+                        $parameters[] = $aggregateHeader->recordedOn;
+                        $types[$offset + 5] = $dateTimeType;
 
-                    $parameters[] = $message->hasHeader(ArchivedHeader::class);
-                    $types[$offset + 7] = $booleanType;
+                        $streamStart = $message->hasHeader(StreamStartHeader::class);
 
-                    $parameters[] = $this->headersSerializer->serialize($this->getCustomHeaders($message));
+                        if ($streamStart) {
+                            $key = $aggregateHeader->aggregateName . '/' . $aggregateHeader->aggregateId;
+                            $achievedUntilPlayhead[$key] = $aggregateHeader->playhead;
+                        }
 
-                    $position++;
+                        $parameters[] = $streamStart;
+                        $types[$offset + 6] = $booleanType;
 
-                    if ($position !== $batchSize) {
-                        continue;
-                    }
+                        $parameters[] = $message->hasHeader(ArchivedHeader::class);
+                        $types[$offset + 7] = $booleanType;
 
-                    $this->executeSave($columns, $placeholders, $parameters, $types, $connection);
+                        $parameters[] = $this->headersSerializer->serialize($this->getCustomHeaders($message));
 
-                    $parameters = [];
-                    $placeholders = [];
-                    $types = [];
+                        $position++;
 
-                    $position = 0;
-                }
+                        if ($position !== $batchSize) {
+                            continue;
+                        }
 
-                if ($position !== 0) {
-                    $this->executeSave($columns, $placeholders, $parameters, $types, $connection);
-                }
+                        $this->executeSave($columns, $placeholders, $parameters, $types, $connection);
 
-                foreach ($achievedUntilPlayhead as $key => $playhead) {
-                    [$aggregateName, $aggregateId] = explode('/', $key);
+                        $parameters = [];
+                        $placeholders = [];
+                        $types = [];
 
-                    $connection->executeStatement(
-                        sprintf(
-                            <<<'SQL'
+                        $position = 0;
+                    }
+
+                    if ($position !== 0) {
+                        $this->executeSave($columns, $placeholders, $parameters, $types, $connection);
+                    }
+
+                    foreach ($achievedUntilPlayhead as $key => $playhead) {
+                        [$aggregateName, $aggregateId] = explode('/', $key);
+
+                        $connection->executeStatement(
+                            sprintf(
+                                <<<'SQL'
                             UPDATE %s
                             SET archived = true
                             WHERE aggregate = :aggregate
                             AND aggregate_id = :aggregate_id
                             AND playhead < :playhead
                             AND archived = false
                             SQL,
-                            $this->config['table_name'],
-                        ),
-                        [
-                            'aggregate' => $aggregateName,
-                            'aggregate_id' => $aggregateId,
-                            'playhead' => $playhead,
-                        ],
-                    );
-                }
-            },
-        );
+                                $this->config['table_name'],
+                            ),
+                            [
+                                'aggregate' => $aggregateName,
+                                'aggregate_id' => $aggregateId,
+                                'playhead' => $playhead,
+                            ],
+                        );
+                    }
+                },
+            );
+        } finally {
+            $this->unlock();
+        }
     }
 
     /**
@@ -423,4 +439,79 @@ private function executeSave(
             throw new UniqueConstraintViolation($e);
         }
     }
+
+    private function lock(): void
+    {
+        if (!$this->config['locking']) {
+            return;
+        }
+
+        $platform = $this->connection->getDatabasePlatform();
+
+        if ($platform instanceof PostgreSQLPlatform) {
+            $this->connection->executeStatement(
+                sprintf(
+                    'SELECT pg_advisory_lock(%s)',
+                    $this->config['lock_id'],
+                ),
+            );
+
+            return;
+        }
+
+        if ($platform instanceof MariaDBPlatform || $platform instanceof MySQLPlatform) {
+            $this->connection->fetchAllAssociative(
+                sprintf(
+                    'SELECT GET_LOCK("%s", %d)',
+                    $this->config['lock_id'],
+                    $this->config['lock_timeout'],
+                ),
+            );
+
+            return;
+        }
+
+        if ($platform instanceof SQLitePlatform) {
+            return; // locking is not supported
+        }
+
+        throw new LockingNotImplemented($platform::class);
+    }
+
+    private function unlock(): void
+    {
+        if (!$this->config['locking']) {
+            return;
+        }
+
+        $platform = $this->connection->getDatabasePlatform();
+
+        if ($platform instanceof PostgreSQLPlatform) {
+            $this->connection->executeStatement(
+                sprintf(
+                    'SELECT pg_advisory_unlock(%s)',
+                    $this->config['lock_id'],
+                ),
+            );
+
+            return;
+        }
+
+        if ($platform instanceof MariaDBPlatform || $platform instanceof MySQLPlatform) {
+            $this->connection->fetchAllAssociative(
+                sprintf(
+                    'SELECT RELEASE_LOCK("%s")',
+                    $this->config['lock_id'],
+                ),
+            );
+
+            return;
+        }
+
+        if ($platform instanceof SQLitePlatform) {
+            return; // locking is not supported
+        }
+
+        throw new LockingNotImplemented($platform::class);
+    }
 }
diff --git a/src/Store/LockingNotImplemented.php b/src/Store/LockingNotImplemented.php
@@ -0,0 +1,21 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Patchlevel\EventSourcing\Store;
+
+use function sprintf;
+
+final class LockingNotImplemented extends StoreException
+{
+    /** @param class-string $platform */
+    public function __construct(string $platform)
+    {
+        parent::__construct(
+            sprintf(
+                'Locking is not implemented on platform %s. Disable locking in the store options.',
+                $platform,
+            ),
+        );
+    }
+}
diff --git a/tests/Unit/Store/DoctrineDbalStoreTest.php b/tests/Unit/Store/DoctrineDbalStoreTest.php
