Skip to content

Commit cf1cb59

Browse files
DanielBaduraDanielBadura
committed
Check the result of GET_LOCK if lock could be acquired and RELEASE_LOCK if we could free it
1 parent 30220c6 commit cf1cb59

11 files changed

Lines changed: 731 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: 26 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -50,6 +50,7 @@
5050
use function is_string;
5151
use function sprintf;
5252

53+
use const PHP_INT_MAX;
5354
use const PHP_VERSION_ID;
5455

5556
final class DoctrineDbalStore implements Store, SubscriptionStore, DoctrineSchemaConfigurator
@@ -493,14 +494,28 @@ private function lock(): void
493494
}
494495

495496
if ($platform instanceof MariaDBPlatform || $platform instanceof MySQLPlatform) {
496-
$this->connection->fetchAllAssociative(
497+
$lockTimeout = $this->config['lock_timeout'];
498+
499+
if ($platform instanceof MariaDBPlatform && $lockTimeout < 0) {
500+
$lockTimeout = PHP_INT_MAX;
501+
}
502+
503+
$result = $this->connection->fetchOne(
497504
sprintf(
498505
'SELECT GET_LOCK("%s", %d)',
499506
$this->config['lock_id'],
500-
$this->config['lock_timeout'],
507+
$lockTimeout,
501508
),
502509
);
503510

511+
if ($result === 0) {
512+
throw LockCouldNotBeAcquired::byTimeout($this->config['lock_id'], $this->config['lock_timeout']);
513+
}
514+
515+
if ($result !== 1) {
516+
throw LockCouldNotBeAcquired::byError($this->config['lock_id']);
517+
}
518+
504519
return;
505520
}
506521

@@ -522,13 +537,21 @@ private function unlock(): void
522537
}
523538

524539
if ($platform instanceof MariaDBPlatform || $platform instanceof MySQLPlatform) {
525-
$this->connection->fetchAllAssociative(
540+
$result = $this->connection->fetchOne(
526541
sprintf(
527542
'SELECT RELEASE_LOCK("%s")',
528543
$this->config['lock_id'],
529544
),
530545
);
531546

547+
if ($result === 0) {
548+
throw LockCouldNotBeFreed::notOurs($this->config['lock_id']);
549+
}
550+
551+
if ($result !== 1) {
552+
throw LockCouldNotBeFreed::notExist($this->config['lock_id']);
553+
}
554+
532555
return;
533556
}
534557

‎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: 26 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -59,6 +59,7 @@
5959
use function str_contains;
6060
use function str_replace;
6161

62+
use const PHP_INT_MAX;
6263
use const PHP_VERSION_ID;
6364

6465
final class StreamDoctrineDbalStore implements StreamStore, SubscriptionStore, DoctrineSchemaConfigurator
@@ -572,14 +573,28 @@ private function lock(): void
572573
}
573574

574575
if ($platform instanceof MariaDBPlatform || $platform instanceof MySQLPlatform) {
575-
$this->connection->fetchAllAssociative(
576+
$lockTimeout = $this->config['lock_timeout'];
577+
578+
if ($platform instanceof MariaDBPlatform && $lockTimeout < 0) {
579+
$lockTimeout = PHP_INT_MAX;
580+
}
581+
582+
$result = $this->connection->fetchOne(
576583
sprintf(
577584
'SELECT GET_LOCK("%s", %d)',
578585
$this->config['lock_id'],
579-
$this->config['lock_timeout'],
586+
$lockTimeout,
580587
),
581588
);
582589

590+
if ($result === 0) {
591+
throw LockCouldNotBeAcquired::byTimeout($this->config['lock_id'], $this->config['lock_timeout']);
592+
}
593+
594+
if ($result !== 1) {
595+
throw LockCouldNotBeAcquired::byError($this->config['lock_id']);
596+
}
597+
583598
return;
584599
}
585600

@@ -601,13 +616,21 @@ private function unlock(): void
601616
}
602617

603618
if ($platform instanceof MariaDBPlatform || $platform instanceof MySQLPlatform) {
604-
$this->connection->fetchAllAssociative(
619+
$result = $this->connection->fetchOne(
605620
sprintf(
606621
'SELECT RELEASE_LOCK("%s")',
607622
$this->config['lock_id'],
608623
),
609624
);
610625

626+
if ($result === 0) {
627+
throw LockCouldNotBeFreed::notOurs($this->config['lock_id']);
628+
}
629+
630+
if ($result !== 1) {
631+
throw LockCouldNotBeFreed::notExist($this->config['lock_id']);
632+
}
633+
611634
return;
612635
}
613636

‎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)