Skip to content

Commit 9b9b482

Browse files
DanielBaduraDanielBadura
authored
Merge pull request #870 from patchlevel/3.19.x-merge-up-into-3.20.x_3xzYlsHZ
2 parents 3f10e46 + 273ad58 commit 9b9b482

11 files changed

Lines changed: 837 additions & 25 deletions

‎docker-compose.yaml‎

Lines changed: 9 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -13,4 +13,12 @@ services:
1313
- MYSQL_ALLOW_EMPTY_PASSWORD="yes"
1414
- MYSQL_DATABASE=eventstore
1515
ports:
16-
- 3306:3306
16+
- 3306:3306
17+
18+
mariadb:
19+
image: mariadb:12
20+
environment:
21+
- MYSQL_ALLOW_EMPTY_PASSWORD="yes"
22+
- MYSQL_DATABASE=eventstore
23+
ports:
24+
- 3307:3306

‎src/Store/DoctrineDbalStore.php‎

Lines changed: 33 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -64,6 +64,14 @@ final class DoctrineDbalStore implements Store, SubscriptionStore, DoctrineSchem
6464
*/
6565
private const DEFAULT_LOCK_ID = 133742;
6666

67+
/**
68+
* MariaDB does not support an infinite (negative) lock timeout. Very large values such as
69+
* PHP_INT_MAX overflow its internal timeout arithmetic and make GET_LOCK return NULL. We
70+
* therefore use a large but safe value (INT32_MAX minus a small buffer) as "effectively
71+
* infinite" wait.
72+
*/
73+
private const INFINITE_MARIADB_LOCK_TIMEOUT = 2_147_482_647;
74+
6775
private readonly HeadersSerializer $headersSerializer;
6876

6977
/** @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
493501
}
494502

495503
if ($platform instanceof MariaDBPlatform || $platform instanceof MySQLPlatform) {
496-
$this->connection->fetchAllAssociative(
504+
$lockTimeout = $this->config['lock_timeout'];
505+
506+
if ($platform instanceof MariaDBPlatform && $lockTimeout < 0) {
507+
$lockTimeout = self::INFINITE_MARIADB_LOCK_TIMEOUT;
508+
}
509+
510+
$result = $this->connection->fetchOne(
497511
sprintf(
498512
'SELECT GET_LOCK("%s", %d)',
499513
$this->config['lock_id'],
500-
$this->config['lock_timeout'],
514+
$lockTimeout,
501515
),
502516
);
503517

518+
if ($result === 0) {
519+
throw LockCouldNotBeAcquired::byTimeout($this->config['lock_id'], $this->config['lock_timeout']);
520+
}
521+
522+
if ($result !== 1) {
523+
throw LockCouldNotBeAcquired::byError($this->config['lock_id']);
524+
}
525+
504526
return;
505527
}
506528

@@ -522,13 +544,21 @@ private function unlock(): void
522544
}
523545

524546
if ($platform instanceof MariaDBPlatform || $platform instanceof MySQLPlatform) {
525-
$this->connection->fetchAllAssociative(
547+
$result = $this->connection->fetchOne(
526548
sprintf(
527549
'SELECT RELEASE_LOCK("%s")',
528550
$this->config['lock_id'],
529551
),
530552
);
531553

554+
if ($result === 0) {
555+
throw LockCouldNotBeFreed::notOurs($this->config['lock_id']);
556+
}
557+
558+
if ($result !== 1) {
559+
throw LockCouldNotBeFreed::notExist($this->config['lock_id']);
560+
}
561+
532562
return;
533563
}
534564

‎src/Store/LockCouldNotBeAcquired.php‎

Lines changed: 24 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,24 @@
1+
<?php
2+
3+
declare(strict_types=1);
4+
5+
namespace Patchlevel\EventSourcing\Store;
6+
7+
use function sprintf;
8+
9+
final class LockCouldNotBeAcquired extends StoreException
10+
{
11+
public static function byTimeout(int $id, int $timeout): self
12+
{
13+
return new self(sprintf(
14+
'The lock with id [%s] could not be acquired with a timeout of %d',
15+
$id,
16+
$timeout,
17+
));
18+
}
19+
20+
public static function byError(int $id): self
21+
{
22+
return new self(sprintf('There was an error when tried to get the lock with id [%s]', $id));
23+
}
24+
}

‎src/Store/LockCouldNotBeFreed.php‎

Lines changed: 20 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,20 @@
1+
<?php
2+
3+
declare(strict_types=1);
4+
5+
namespace Patchlevel\EventSourcing\Store;
6+
7+
use function sprintf;
8+
9+
final class LockCouldNotBeFreed extends StoreException
10+
{
11+
public static function notExist(int $id): self
12+
{
13+
return new self(sprintf('The lock with id [%s] could not be freed as it does not exist', $id));
14+
}
15+
16+
public static function notOurs(int $id): self
17+
{
18+
return new self(sprintf('The lock with id [%s] could not be freed as it is not ours', $id));
19+
}
20+
}

‎src/Store/StreamDoctrineDbalStore.php‎

Lines changed: 33 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -73,6 +73,14 @@ final class StreamDoctrineDbalStore implements StreamStore, SubscriptionStore, D
7373
*/
7474
private const DEFAULT_LOCK_ID = 133742;
7575

76+
/**
77+
* MariaDB does not support an infinite (negative) lock timeout. Very large values such as
78+
* PHP_INT_MAX overflow its internal timeout arithmetic and make GET_LOCK return NULL. We
79+
* therefore use a large but safe value (INT32_MAX minus a small buffer) as "effectively
80+
* infinite" wait.
81+
*/
82+
private const INFINITE_MARIADB_LOCK_TIMEOUT = 2_147_482_647;
83+
7684
private readonly HeadersSerializer $headersSerializer;
7785

7886
private readonly ClockInterface $clock;
@@ -572,14 +580,28 @@ private function lock(): void
572580
}
573581

574582
if ($platform instanceof MariaDBPlatform || $platform instanceof MySQLPlatform) {
575-
$this->connection->fetchAllAssociative(
583+
$lockTimeout = $this->config['lock_timeout'];
584+
585+
if ($platform instanceof MariaDBPlatform && $lockTimeout < 0) {
586+
$lockTimeout = self::INFINITE_MARIADB_LOCK_TIMEOUT;
587+
}
588+
589+
$result = $this->connection->fetchOne(
576590
sprintf(
577591
'SELECT GET_LOCK("%s", %d)',
578592
$this->config['lock_id'],
579-
$this->config['lock_timeout'],
593+
$lockTimeout,
580594
),
581595
);
582596

597+
if ($result === 0) {
598+
throw LockCouldNotBeAcquired::byTimeout($this->config['lock_id'], $this->config['lock_timeout']);
599+
}
600+
601+
if ($result !== 1) {
602+
throw LockCouldNotBeAcquired::byError($this->config['lock_id']);
603+
}
604+
583605
return;
584606
}
585607

@@ -601,13 +623,21 @@ private function unlock(): void
601623
}
602624

603625
if ($platform instanceof MariaDBPlatform || $platform instanceof MySQLPlatform) {
604-
$this->connection->fetchAllAssociative(
626+
$result = $this->connection->fetchOne(
605627
sprintf(
606628
'SELECT RELEASE_LOCK("%s")',
607629
$this->config['lock_id'],
608630
),
609631
);
610632

633+
if ($result === 0) {
634+
throw LockCouldNotBeFreed::notOurs($this->config['lock_id']);
635+
}
636+
637+
if ($result !== 1) {
638+
throw LockCouldNotBeFreed::notExist($this->config['lock_id']);
639+
}
640+
611641
return;
612642
}
613643

‎tests/Integration/Store/DoctrineDbalStoreTest.php‎

Lines changed: 48 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -6,12 +6,19 @@
66

77
use DateTimeImmutable;
88
use Doctrine\DBAL\Connection;
9+
use Doctrine\DBAL\DriverManager;
10+
use Doctrine\DBAL\Platforms\PostgreSQLPlatform;
11+
use Doctrine\DBAL\Platforms\SQLitePlatform;
912
use Doctrine\DBAL\Schema\Schema;
1013
use Patchlevel\EventSourcing\Aggregate\AggregateHeader;
1114
use Patchlevel\EventSourcing\Message\Message;
1215
use Patchlevel\EventSourcing\Schema\DoctrineSchemaDirector;
1316
use Patchlevel\EventSourcing\Serializer\DefaultEventSerializer;
1417
use Patchlevel\EventSourcing\Store\DoctrineDbalStore;
18+
use Patchlevel\EventSourcing\Store\Header\PlayheadHeader;
19+
use Patchlevel\EventSourcing\Store\Header\RecordedOnHeader;
20+
use Patchlevel\EventSourcing\Store\Header\StreamNameHeader;
21+
use Patchlevel\EventSourcing\Store\LockCouldNotBeAcquired;
1522
use Patchlevel\EventSourcing\Store\Store;
1623
use Patchlevel\EventSourcing\Store\UniqueConstraintViolation;
1724
use Patchlevel\EventSourcing\Tests\DbalManager;
@@ -20,6 +27,7 @@
2027
use PHPUnit\Framework\TestCase;
2128

2229
use function json_decode;
30+
use function sprintf;
2331

2432
#[CoversNothing]
2533
final class DoctrineDbalStoreTest extends TestCase
@@ -34,6 +42,7 @@ public function setUp(): void
3442
$this->store = new DoctrineDbalStore(
3543
$this->connection,
3644
DefaultEventSerializer::createFromPaths([__DIR__ . '/Events']),
45+
config: ['lock_timeout' => 1],
3746
);
3847

3948
$schemaDirector = new DoctrineSchemaDirector(
@@ -190,6 +199,45 @@ public function testSave10000Messages(): void
190199
self::assertEquals(10000, $result);
191200
}
192201

202+
public function testSaveLockTimeout(): void
203+
{
204+
if ($this->connection->getDatabasePlatform() instanceof SQLitePlatform) {
205+
$this->markTestSkipped('SQLite does not support locks');
206+
}
207+
208+
if ($this->connection->getDatabasePlatform() instanceof PostgreSQLPlatform) {
209+
$this->markTestSkipped('PostgreSQL does lock indefinitely');
210+
}
211+
212+
$profileId = ProfileId::generate();
213+
214+
$messages = [
215+
Message::create(new ProfileCreated($profileId, 'test'))
216+
->withHeader(new StreamNameHeader(sprintf('profile-%s', $profileId->toString())))
217+
->withHeader(new PlayheadHeader(1))
218+
->withHeader(new RecordedOnHeader(new DateTimeImmutable('2020-01-01 00:00:00'))),
219+
];
220+
221+
$connection = DriverManager::getConnection($this->connection->getParams());
222+
223+
$lock = $connection->fetchOne(
224+
sprintf(
225+
'SELECT GET_LOCK("%s", %d)',
226+
133742,
227+
1,
228+
),
229+
);
230+
self::assertSame(1, $lock);
231+
232+
$this->expectException(LockCouldNotBeAcquired::class);
233+
$this->expectExceptionMessage('The lock with id [133742] could not be acquired with a timeout of 1');
234+
try {
235+
$this->store->save(...$messages);
236+
} finally {
237+
$connection->close();
238+
}
239+
}
240+
193241
public function testLoad(): void
194242
{
195243
$profileId = ProfileId::generate();

‎tests/Integration/Store/StreamDoctrineDbalStoreTest.php‎

Lines changed: 44 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -6,6 +6,9 @@
66

77
use DateTimeImmutable;
88
use Doctrine\DBAL\Connection;
9+
use Doctrine\DBAL\DriverManager;
10+
use Doctrine\DBAL\Platforms\PostgreSQLPlatform;
11+
use Doctrine\DBAL\Platforms\SQLitePlatform;
912
use Doctrine\DBAL\Schema\Schema;
1013
use Patchlevel\EventSourcing\Clock\FrozenClock;
1114
use Patchlevel\EventSourcing\Message\Message;
@@ -19,6 +22,7 @@
1922
use Patchlevel\EventSourcing\Store\Header\PlayheadHeader;
2023
use Patchlevel\EventSourcing\Store\Header\RecordedOnHeader;
2124
use Patchlevel\EventSourcing\Store\Header\StreamNameHeader;
25+
use Patchlevel\EventSourcing\Store\LockCouldNotBeAcquired;
2226
use Patchlevel\EventSourcing\Store\StreamDoctrineDbalStore;
2327
use Patchlevel\EventSourcing\Store\StreamStore;
2428
use Patchlevel\EventSourcing\Store\UniqueConstraintViolation;
@@ -51,6 +55,7 @@ public function setUp(): void
5155
$this->connection,
5256
DefaultEventSerializer::createFromPaths([__DIR__ . '/Events']),
5357
clock: $this->clock,
58+
config: ['lock_timeout' => 1],
5459
);
5560

5661
$schemaDirector = new DoctrineSchemaDirector(
@@ -396,6 +401,45 @@ public function testSave10000Messages(): void
396401
self::assertEquals(10000, $result);
397402
}
398403

404+
public function testSaveLockTimeout(): void
405+
{
406+
if ($this->connection->getDatabasePlatform() instanceof SQLitePlatform) {
407+
$this->markTestSkipped('SQLite does not support locks');
408+
}
409+
410+
if ($this->connection->getDatabasePlatform() instanceof PostgreSQLPlatform) {
411+
$this->markTestSkipped('PostgreSQL does lock indefinitely');
412+
}
413+
414+
$profileId = ProfileId::generate();
415+
416+
$messages = [
417+
Message::create(new ProfileCreated($profileId, 'test'))
418+
->withHeader(new StreamNameHeader(sprintf('profile-%s', $profileId->toString())))
419+
->withHeader(new PlayheadHeader(1))
420+
->withHeader(new RecordedOnHeader(new DateTimeImmutable('2020-01-01 00:00:00'))),
421+
];
422+
423+
$connection = DriverManager::getConnection($this->connection->getParams());
424+
425+
$lock = $connection->fetchOne(
426+
sprintf(
427+
'SELECT GET_LOCK("%s", %d)',
428+
133742,
429+
1,
430+
),
431+
);
432+
self::assertSame(1, $lock);
433+
434+
$this->expectException(LockCouldNotBeAcquired::class);
435+
$this->expectExceptionMessage('The lock with id [133742] could not be acquired with a timeout of 1');
436+
try {
437+
$this->store->save(...$messages);
438+
} finally {
439+
$connection->close();
440+
}
441+
}
442+
399443
public function testLoad(): void
400444
{
401445
$profileId = ProfileId::generate();

0 commit comments

Comments
 (0)