diff --git a/Classes/Domain/AbstractScheduler.php b/Classes/Domain/AbstractScheduler.php index 7d1b0e0..42a3a4c 100644 --- a/Classes/Domain/AbstractScheduler.php +++ b/Classes/Domain/AbstractScheduler.php @@ -6,15 +6,14 @@ use DateTimeImmutable; use Doctrine\DBAL\Exception; -use Doctrine\DBAL\Exception\RetryableException; use Doctrine\DBAL\Types\Types; use InvalidArgumentException; use Neos\Flow\Utility\Algorithms; use Netlogix\JobQueue\Scheduled\Domain\Model\ScheduledJob; use Netlogix\JobQueue\Scheduled\DueDateCalculation\TimeBaseForDueDateCalculation; use Netlogix\JobQueue\Scheduled\Service\Connection; -use Netlogix\Retry\Retry; use Neos\Flow\Annotations as Flow; +use Throwable; use function array_filter; use function in_array; @@ -51,8 +50,8 @@ public function injectConnection(Connection $connection): void $this->dbal = $connection; } - public function injectTimeBaseForDueDateCalculation(TimeBaseForDueDateCalculation $timeBaseForDueDateCalculation): void - { + public function injectTimeBaseForDueDateCalculation(TimeBaseForDueDateCalculation $timeBaseForDueDateCalculation + ): void { $this->timeBaseForDueDateCalculation = $timeBaseForDueDateCalculation; } @@ -99,37 +98,36 @@ public function next(string $groupName): ?ScheduledJob $claim = Algorithms::generateUUID(); $claimQuery = static::CLAIM_QUERY; - (new Retry()) - /** - * @see http://backoffcalculator.com/?attempts=5&rate=1&interval=0.5 - */ - ->withExponentialBackoff(retryInterval: 0.5, maxRetries: 5) - ->onExceptionsOfType(RetryableException::class) - ->task(fn() => $this->dbal - ->executeQuery( - $claimQuery, - [ - 'now' => $this->timeBaseForDueDateCalculation->getNow(), - 'groupname' => $groupName, - 'claimed' => $claim, - ], - [ - 'now' => Types::DATETIME_IMMUTABLE, - 'groupname' => Types::STRING, - 'claimed' => Types::STRING, - ] - )); + $this->dbal + ->executeQuery( + sql: $claimQuery, + params: [ + 'now' => $this->timeBaseForDueDateCalculation->getNow(), + 'groupname' => $groupName, + 'claimed' => $claim, + ], + types: [ + 'now' => Types::DATETIME_IMMUTABLE, + 'groupname' => Types::STRING, + 'claimed' => Types::STRING, + ], + logContext: fn (Throwable $throwable, int $incarnation) => [ + 'claim' => $claim, + 'groupName' => $groupName, + 'step' => 'claim', + ] + ); $select = static::SELECT_QUERY; $row = $this->dbal ->executeQuery( - $select, - [ + sql: $select, + params: [ 'groupname' => $groupName, 'claimed' => $claim, ], - [ + types: [ 'groupname' => Types::STRING, 'claimed' => Types::STRING, ] @@ -142,34 +140,33 @@ public function next(string $groupName): ?ScheduledJob $release = static::RELEASE_QUERY; - (new Retry()) - /** - * @see http://backoffcalculator.com/?attempts=5&rate=1&interval=0.5 - */ - ->withExponentialBackoff(retryInterval: 0.5, maxRetries: 5) - ->onExceptionsOfType(RetryableException::class) - ->task(fn() => $this->dbal - ->executeQuery( - $release, - [ - 'groupname' => $groupName, - 'claimed' => $claim, - ], - [ - 'groupname' => Types::STRING, - 'claimed' => Types::STRING, - ] - )); + $this->dbal + ->executeQuery( + sql: $release, + params: [ + 'groupname' => $groupName, + 'claimed' => $claim, + ], + types: [ + 'groupname' => Types::STRING, + 'claimed' => Types::STRING, + ], + logContext: fn (Throwable $throwable, int $incarnation) => [ + 'claim' => $claim, + 'groupName' => $groupName, + 'step' => 'release', + ] + ); return ScheduledJob::createInternal( job: $row['job'], queue: $row['queue'], duedate: new DateTimeImmutable($row['duedate']), groupName: $groupName, - identifier: (string)$row['identifier'], - incarnation: (int)$row['incarnation'], - claimed: (string)$row['claimed'], - running: (int)$row['running'] + identifier: (string) $row['identifier'], + incarnation: (int) $row['incarnation'], + claimed: (string) $row['claimed'], + running: (int) $row['running'] ); } @@ -180,34 +177,36 @@ public function release(ScheduledJob $job): void } $tableName = ScheduledJob::TABLE_NAME; - $delete = /** @lang MySQL */ <<<"MySQL" - DELETE FROM {$tableName} - WHERE groupname = :groupname - AND identifier = :identifier - AND claimed = :claimed - MySQL; + $delete = /** @lang MySQL */ + <<<"MySQL" + DELETE FROM {$tableName} + WHERE groupname = :groupname + AND identifier = :identifier + AND claimed = :claimed + MySQL; $deleteResult = $this->dbal ->executeQuery( - $delete, - [ + sql: $delete, + params: [ 'groupname' => $job->getGroupName(), 'identifier' => $job->getIdentifier(), 'claimed' => $job->getClaimed(), ] ); if ($deleteResult->rowCount() === 0) { - $free = /** @lang MySQL */ <<dbal ->executeQuery( - $free, - [ + sql: $free, + params: [ 'groupname' => $job->getGroupName(), 'identifier' => $job->getIdentifier(), ] @@ -225,22 +224,23 @@ public function fail(ScheduledJob $job, string $reason): void $tableName = ScheduledJob::TABLE_NAME; - $update = /** @lang MySQL */ <<dbal ->executeQuery( - $update, - [ + sql: $update, + params: [ 'identifier' => $job->getIdentifier(), 'claimed' => $job->getClaimed(), 'failed' => sprintf('failed(%s)', $reason), ], - [ + types: [ 'identifier' => Types::STRING, 'claimed' => Types::STRING, 'failed' => Types::STRING, @@ -256,24 +256,27 @@ public function fail(ScheduledJob $job, string $reason): void * @return void * @Flow\Signal */ - public function emitFailed(ScheduledJob $job, string $reason): void {} + public function emitFailed(ScheduledJob $job, string $reason): void + { + } public function activity(ScheduledJob $job): void { $tableName = ScheduledJob::TABLE_NAME; - $update = /** @lang MySQL */ <<<"MySQL" - UPDATE {$tableName} - SET activity = NOW() - WHERE identifier = :identifier - MySQL; + $update = /** @lang MySQL */ + <<<"MySQL" + UPDATE {$tableName} + SET activity = NOW() + WHERE identifier = :identifier + MySQL; $this->dbal ->executeQuery( - $update, - [ + sql: $update, + params: [ 'identifier' => $job->getIdentifier(), ], - [ + types: [ 'identifier' => Types::STRING, ] ); @@ -284,24 +287,25 @@ public function activity(ScheduledJob $job): void * * @param string $groupName Free jobs in this group only * @param int|null $minutes Only free jobs that are stale for at least this many minutes. @deprecated Use staleJobTimeout configuration setting instead. - * @throws Exception * @return int Number of freed jobs + * @throws Exception */ public function resetStaleJobs( string $groupName, ?int $minutes = null ): int { - return $this->dbal->executeQuery( - sql: static::RESET_STALE_JOBS_QUERY, - params: [ - 'groupName' => $groupName, - 'seconds' => max($minutes === null ? $this->staleJobTimeoutSecs : $minutes * 60, 1), - ], - types: [ - 'groupName' => Types::STRING, - 'minutes' => Types::SMALLINT, - ], - )->rowCount(); + return $this->dbal + ->executeQuery( + sql: static::RESET_STALE_JOBS_QUERY, + params: [ + 'groupName' => $groupName, + 'seconds' => max($minutes === null ? $this->staleJobTimeoutSecs : $minutes * 60, 1), + ], + types: [ + 'groupName' => Types::STRING, + 'seconds' => Types::SMALLINT, + ], + )->rowCount(); } protected function scheduleJob(ScheduledJob $job): void @@ -309,36 +313,34 @@ protected function scheduleJob(ScheduledJob $job): void $this->validateGroupName($job->getGroupName()); $statement = static::SCHEDULE_QUERY; - (new Retry()) - /** - * @see http://backoffcalculator.com/?attempts=5&rate=1&interval=0.1 - */ - ->withExponentialBackoff(retryInterval: 0.05, maxRetries: 5) - ->onExceptionsOfType(RetryableException::class) - ->task(fn() => $this->dbal - ->executeQuery( - $statement, - [ - 'groupname' => $job->getGroupName(), - 'identifier' => $job->getIdentifier(), - 'duedate' => $job->getDuedate(), - 'queue' => $job->getQueueName(), - 'job' => $job->getSerializedJob(), - 'incarnation' => $job->getIncarnation(), - 'claimed' => $job->getClaimed(), - 'running' => $job->getRunning(), - ], - [ - 'groupname' => Types::STRING, - 'identifier' => Types::STRING, - 'duedate' => Types::DATETIME_IMMUTABLE, - 'queue' => Types::STRING, - 'job' => Types::BLOB, - 'incarnation' => Types::INTEGER, - 'claimed' => Types::STRING, - 'running' => Types::INTEGER, - ] - )); + $this->dbal + ->executeQuery( + sql: $statement, + params: [ + 'groupname' => $job->getGroupName(), + 'identifier' => $job->getIdentifier(), + 'duedate' => $job->getDuedate(), + 'queue' => $job->getQueueName(), + 'job' => $job->getSerializedJob(), + 'incarnation' => $job->getIncarnation(), + 'claimed' => $job->getClaimed(), + 'running' => $job->getRunning(), + ], + types: [ + 'groupname' => Types::STRING, + 'identifier' => Types::STRING, + 'duedate' => Types::DATETIME_IMMUTABLE, + 'queue' => Types::STRING, + 'job' => Types::BLOB, + 'incarnation' => Types::INTEGER, + 'claimed' => Types::STRING, + 'running' => Types::INTEGER, + ], + logContext: fn (Throwable $throwable, int $incarnation) => [ + 'groupName' => $job->getGroupName(), + 'step' => 'schedule', + ] + ); // TODO: Find a way to "trigger queueing" without cronjobs. Maybe "dynamic cronjobs" like "at". // TODO: On Shutdown: Add queueing job to job queue. } @@ -350,11 +352,13 @@ protected function validateGroupName(string $groupName): void } } - public function getConnection(): Connection { + public function getConnection(): Connection + { return $this->dbal; } - public function getStaleJobTimeoutSeconds(): int { + public function getStaleJobTimeoutSeconds(): int + { return $this->staleJobTimeoutSecs; } } diff --git a/Classes/Domain/PostgreSQLScheduler.php b/Classes/Domain/PostgreSQLScheduler.php index d59b688..f9377bf 100644 --- a/Classes/Domain/PostgreSQLScheduler.php +++ b/Classes/Domain/PostgreSQLScheduler.php @@ -5,14 +5,25 @@ class PostgreSQLScheduler extends AbstractScheduler { /** + * The candidate row MUST be selected in a `MATERIALIZED` CTE, never in an + * inline `FROM (SELECT ... LIMIT 1 FOR UPDATE SKIP LOCKED)` subquery. + * + * PostgreSQL is free to place such an inline subquery on the inner side of + * a nested-loop join and re-evaluate it once per outer row. Combined with + * `FOR UPDATE SKIP LOCKED`, every re-evaluation skips the rows already + * locked by previous iterations and returns the *next* candidate, so the + * join ends up matching - and claiming - more than the single intended row + * (all with the same claim value). Whether this happens depends on the + * query plan, i.e. on table size and statistics, which is why it only + * surfaces on large production tables and not on small dev databases. + * + * `AS MATERIALIZED` forces the candidate selection to be evaluated exactly + * once, so `LIMIT 1` reliably bounds the update to a single row. + * * @lang PostgreSQL */ protected const CLAIM_QUERY = <<throwableStorage = $throwableStorage; + } + /** * Use the same database credentials as the entity manager but create * a new connection. All SQL queries issued by the scheduler are meant @@ -43,16 +63,20 @@ public function injectEntityManager(EntityManagerInterface $entityManager): void $this->dbal->connect(); } - public function fetchOne(string $query, array $params = [], array $types = []) + public function fetchOne(string $query, array $params = [], array $types = [], ?callable $logContext = null) { return $this->withAutoReconnectAndRetry(function () use ($query, $params, $types) { return $this->dbal->fetchOne($query, $params, $types); - }); + }, logContext: $logContext); } - public function fetchOneReadUncommited(string $query, array $params = [], array $types = []) - { - return $this->withAutoReconnectAndRetry(function () use ($query, $params, $types) { + public function fetchOneReadUncommited( + string $query, + array $params = [], + array $types = [], + ?callable $logContext = null + ) { + return $this->withAutoReconnectAndRetry(dbalInteraction: function () use ($query, $params, $types) { $previous = $this->dbal->getTransactionIsolation(); try { $this->dbal->setTransactionIsolation(TransactionIsolationLevel::READ_UNCOMMITTED); @@ -62,14 +86,27 @@ public function fetchOneReadUncommited(string $query, array $params = [], array } finally { $this->dbal->setTransactionIsolation($previous); } - }); + }, logContext: $logContext); } - public function executeQuery($sql, array $params = [], $types = [], ?QueryCacheProfile $qcp = null) - { + /** + * @param string $sql + * @param array $params + * @param array $types + * @param QueryCacheProfile|null $qcp + * @param null|callable(Throwable $throwable, int $incarnation): array $logContext + * @return mixed + */ + public function executeQuery( + string $sql, + array $params = [], + array $types = [], + ?QueryCacheProfile $qcp = null, + ?callable $logContext = null + ) { return $this->withAutoReconnectAndRetry(function () use ($sql, $params, $types, $qcp) { return $this->dbal->executeQuery($sql, $params, $types, $qcp); - }); + }, logContext: $logContext); } public function ping(): void @@ -83,20 +120,49 @@ public function ping(): void * are meant to be atomic anyway, there should be no lost data and no data * duplication. * + * RetryableExceptions (deadlocks, lock wait timeouts, …) and lost + * connections are retried with exponential backoff, and every failure that + * is followed by another attempt is logged. The final, exhausted failure is + * rethrown to the caller instead (and logged there). The optional + * $logContext callable may enrich the data that is logged for every + * retried throwable, e.g. to add the current step, claim or group name. + * * @template T * @param callable(): T $dbalInteraction + * @param null|callable(Throwable $throwable, int $incarnation): array $logContext * @return T */ - protected function withAutoReconnectAndRetry(callable $dbalInteraction) + protected function withAutoReconnectAndRetry(callable $dbalInteraction, ?callable $logContext = null) { - try { - return $dbalInteraction(); - } catch (ConnectionLost) { - $this->dbal->connect(); - return $dbalInteraction(); - } catch (RetryableException) { - return $dbalInteraction(); - } + $maxRetries = $this->maxRetries; + + return (new Retry()) + ->withExponentialBackoff(retryInterval: $this->retryInterval, maxRetries: $maxRetries) + ->onExceptionsOfType(RetryableException::class, ConnectionLost::class) + ->onError(function (Throwable $throwable, int $incarnation, bool $shouldConsider) use ($logContext, $maxRetries) { + if (!$shouldConsider) { + return; + } + if ($incarnation >= $maxRetries) { + // The retry budget is exhausted, so this throwable will be + // rethrown to the caller and logged upstream. There is no + // next attempt to prepare or log for. + return; + } + if ($throwable instanceof ConnectionLost) { + // Force a fresh connection before the next attempt. A bare + // connect() is a no-op while DBAL still holds the stale + // handle, so close() first. + $this->dbal->close(); + $this->dbal->connect(); + } + $additionalData = ['incarnation' => $incarnation]; + if ($logContext !== null) { + $additionalData = [...$additionalData, ...$logContext($throwable, $incarnation)]; + } + $this->throwableStorage->logThrowable($throwable, $additionalData); + }) + ->task($dbalInteraction); } /** @@ -115,7 +181,8 @@ protected function withAutoReconnect(callable $dbalInteraction) return $this->withAutoReconnectAndRetry($dbalInteraction); } - public function getDbal(): DBALConnection { + public function getDbal(): DBALConnection + { return $this->dbal; } } diff --git a/Tests/Functional/RetrievingTest.php b/Tests/Functional/RetrievingTest.php index c88ddf9..420bc94 100644 --- a/Tests/Functional/RetrievingTest.php +++ b/Tests/Functional/RetrievingTest.php @@ -4,13 +4,15 @@ namespace Netlogix\JobQueue\Scheduled\Tests\Functional; use DateTimeImmutable; +use Doctrine\DBAL\Connection as DBALConnection; use Doctrine\DBAL\Exception\DeadlockException; use Doctrine\ORM\EntityManagerInterface; +use Neos\Flow\Log\ThrowableStorageInterface; use Neos\Flow\Persistence\Doctrine\PersistenceManager; use Neos\Flow\Persistence\PersistenceManagerInterface; use Netlogix\JobQueue\Scheduled\Domain\Model\ScheduledJob; use Netlogix\JobQueue\Scheduled\Domain\Scheduler; -use Netlogix\JobQueue\Scheduled\Service\Connection; +use Netlogix\JobQueue\Scheduled\Tests\Functional\Service\TestableConnection; use function serialize; @@ -130,26 +132,20 @@ public function Scheduled_jobs_can_be_retrieved_only_once(): void */ public function Retry_claiming_when_deadlock_exceptions_happen(): void { - $connection = self::createMock(Connection::class); - $connection->expects(self::any()) + // Retrying is the Connection's responsibility (see ConnectionTest), + // so we drive the real retry over a DBAL connection that always + // deadlocks instead of mocking the Connection away. The claim query is + // attempted once plus one per retry, then the exception propagates. + $dbal = self::createMock(DBALConnection::class); + $dbal->expects(self::exactly(TestableConnection::MAX_RETRIES + 1)) ->method('executeQuery') ->willThrowException(self::createStub(DeadlockException::class)); + $connection = new TestableConnection($dbal, self::createMock(ThrowableStorageInterface::class)); $this->scheduler->injectConnection($connection); - $start = microtime(true); - try { - $this->scheduler->next(Scheduler::DEFAULT_GROUP_NAME); - } catch (DeadlockException $e) { - } - $end = microtime(true); - $delta = $end - $start; - - self::assertInstanceOf(DeadlockException::class, $e); - - // guesstimated value is 15 and something. - self::assertGreaterThan(14, $delta); - self::assertLessThan(18, $delta); + $this->expectException(DeadlockException::class); + $this->scheduler->next(Scheduler::DEFAULT_GROUP_NAME); } /** @@ -171,4 +167,40 @@ public function Scheduled_jobs_are_claimed(): void assert($retrievedJob instanceof ScheduledJob); self::assertNotEquals('', $retrievedJob->getClaimed()); } + + /** + * A single claim must mark exactly one row, even when several jobs are due. + * + * The PostgreSQL claim historically used an inline + * "FROM (SELECT ... LIMIT 1 FOR UPDATE SKIP LOCKED)" subquery. The planner + * may put that subquery on the inner side of a nested-loop join and + * re-evaluate it per row; with SKIP LOCKED each re-evaluation returns the + * next candidate, so multiple rows get claimed at once with the same claim + * value. Whether this happens is plan (i.e. table-size) dependent, so this + * test documents the invariant rather than reliably reproducing the plan. + * + * @test + */ + public function A_single_claim_marks_exactly_one_row(): void + { + foreach (['id-1', 'id-2', 'id-3'] as $identifier) { + $this->scheduler->schedule( + ScheduledJob::createNew( + job: self::getJobQueueJob(), + queue: self::getQueueName(), + duedate: $this->now->modify('- 1 day'), + groupName: Scheduler::DEFAULT_GROUP_NAME, + identifier: $identifier + ) + ); + } + + $this->scheduler->next(Scheduler::DEFAULT_GROUP_NAME); + + $claimedRows = (int)$this->scheduler->getConnection()->fetchOne( + 'SELECT COUNT(*) FROM ' . ScheduledJob::TABLE_NAME . " WHERE claimed <> ''" + ); + + self::assertSame(1, $claimedRows, 'A single claim must mark exactly one row'); + } } diff --git a/Tests/Functional/SchedulingTest.php b/Tests/Functional/SchedulingTest.php index a1fb25e..0db51dd 100644 --- a/Tests/Functional/SchedulingTest.php +++ b/Tests/Functional/SchedulingTest.php @@ -3,10 +3,12 @@ namespace Netlogix\JobQueue\Scheduled\Tests\Functional; +use Doctrine\DBAL\Connection as DBALConnection; use Doctrine\DBAL\Exception\DeadlockException; +use Neos\Flow\Log\ThrowableStorageInterface; use Netlogix\JobQueue\Scheduled\Domain\Model\ScheduledJob; use Netlogix\JobQueue\Scheduled\Domain\Scheduler; -use Netlogix\JobQueue\Scheduled\Service\Connection; +use Netlogix\JobQueue\Scheduled\Tests\Functional\Service\TestableConnection; class SchedulingTest extends TestCase { @@ -112,32 +114,26 @@ public function Scheduled_jobs_contain_due_dates(): void */ public function Scheduling_jobs_retries_RetryableExceptions(): void { - $connection = self::createMock(Connection::class); - $connection->expects(self::any()) + // Retrying is the Connection's responsibility (see ConnectionTest), so + // we drive the real retry over a DBAL connection that always deadlocks + // instead of mocking the Connection away. The schedule query is + // attempted once plus one per retry, then the exception propagates. + $dbal = self::createMock(DBALConnection::class); + $dbal->expects(self::exactly(TestableConnection::MAX_RETRIES + 1)) ->method('executeQuery') ->willThrowException(self::createStub(DeadlockException::class)); + $connection = new TestableConnection($dbal, self::createMock(ThrowableStorageInterface::class)); $this->scheduler->injectConnection($connection); - $start = microtime(true); - try { - $this->scheduler->schedule( - ScheduledJob::createNew( - job: self::getJobQueueJob(), - queue: self::getQueueName(), - duedate: self::getDueDate(), - groupName: Scheduler::DEFAULT_GROUP_NAME - ) - ); - } catch (DeadlockException $e) { - } - $end = microtime(true); - $delta = $end - $start; - - self::assertInstanceOf(DeadlockException::class, $e); - - // guesstimated value is about 1.5 - self::assertGreaterThan(1, $delta); - self::assertLessThan(2, $delta); + $this->expectException(DeadlockException::class); + $this->scheduler->schedule( + ScheduledJob::createNew( + job: self::getJobQueueJob(), + queue: self::getQueueName(), + duedate: self::getDueDate(), + groupName: Scheduler::DEFAULT_GROUP_NAME + ) + ); } } diff --git a/Tests/Functional/Service/ConnectionTest.php b/Tests/Functional/Service/ConnectionTest.php new file mode 100644 index 0000000..e5001c3 --- /dev/null +++ b/Tests/Functional/Service/ConnectionTest.php @@ -0,0 +1,211 @@ +dbal = $this->createMock(DBALConnection::class); + $this->throwableStorage = $this->createMock(ThrowableStorageInterface::class); + $this->connection = new TestableConnection($this->dbal, $this->throwableStorage); + } + + protected function tearDown(): void + { + try { + parent::tearDown(); + } catch (\Error $error) { + // FunctionalTestCase::tearDown() fires the "allObjectsPersisted" + // signal, which in this platform is wired to the Neos + // ContentRepository search indexer. Without an ElasticSearch + // backend (as in the test environment) it throws + // "flush() on null", and the base tearDown only catches + // \Exception, not \Error. This test persists no domain objects, so + // the signal is pure noise here. + if (!str_contains($error->getMessage(), 'flush() on null')) { + throw $error; + } + } + } + + /** + * @test + */ + public function Successful_interactions_are_not_retried_and_nothing_is_logged(): void + { + $this->throwableStorage->expects(self::never())->method('logThrowable'); + + $calls = 0; + $result = $this->connection->run(function () use (&$calls) { + $calls++; + return 'ok'; + }); + + self::assertSame('ok', $result); + self::assertSame(1, $calls); + } + + /** + * @test + */ + public function Retryable_exceptions_are_retried_and_logged_until_success(): void + { + $this->throwableStorage->expects(self::once())->method('logThrowable'); + + $exception = $this->retryableException(); + $calls = 0; + $result = $this->connection->run(function () use (&$calls, $exception) { + $calls++; + if ($calls === 1) { + throw $exception; + } + return 'ok'; + }); + + self::assertSame('ok', $result); + self::assertSame(2, $calls); + } + + /** + * @test + * + * Verifies point 1: the final, exhausted attempt is rethrown to the caller + * but is NOT logged again here (it would be logged 6 times otherwise). + */ + public function Exhausted_retryable_exceptions_are_rethrown_and_the_final_attempt_is_not_logged(): void + { + $this->throwableStorage->expects(self::exactly(TestableConnection::MAX_RETRIES))->method('logThrowable'); + + $exception = $this->retryableException(); + $calls = 0; + try { + $this->connection->run(function () use (&$calls, $exception) { + $calls++; + throw $exception; + }); + self::fail('Expected the exhausted retryable exception to be rethrown.'); + } catch (Throwable $caught) { + self::assertSame($exception, $caught); + } + + self::assertSame(TestableConnection::MAX_RETRIES + 1, $calls); + } + + /** + * @test + * + * Verifies point 2: a lost connection is fully re-established (close() then + * connect()) before the next attempt. + */ + public function Connection_lost_triggers_close_and_connect_before_the_next_attempt(): void + { + $this->dbal->expects(self::once())->method('close'); + $this->dbal->expects(self::once())->method('connect'); + $this->throwableStorage->expects(self::once())->method('logThrowable'); + + $exception = $this->connectionLostException(); + $calls = 0; + $result = $this->connection->run(function () use (&$calls, $exception) { + $calls++; + if ($calls === 1) { + throw $exception; + } + return 'ok'; + }); + + self::assertSame('ok', $result); + self::assertSame(2, $calls); + } + + /** + * @test + */ + public function Log_context_is_merged_into_the_logged_additional_data(): void + { + $captured = null; + $this->throwableStorage + ->expects(self::once()) + ->method('logThrowable') + ->willReturnCallback(function (Throwable $throwable, array $additionalData) use (&$captured) { + $captured = $additionalData; + return ''; + }); + + $exception = $this->retryableException(); + $calls = 0; + $this->connection->run( + function () use (&$calls, $exception) { + $calls++; + if ($calls === 1) { + throw $exception; + } + return 'ok'; + }, + fn (Throwable $throwable, int $incarnation) => ['step' => 'claim', 'groupName' => 'some-group'] + ); + + self::assertSame(0, $captured['incarnation']); + self::assertSame('claim', $captured['step']); + self::assertSame('some-group', $captured['groupName']); + } + + /** + * @test + */ + public function Non_retryable_exceptions_are_rethrown_immediately_without_logging(): void + { + $this->throwableStorage->expects(self::never())->method('logThrowable'); + $this->dbal->expects(self::never())->method('close'); + + $exception = new RuntimeException('not retryable'); + $calls = 0; + try { + $this->connection->run(function () use (&$calls, $exception) { + $calls++; + throw $exception; + }); + self::fail('Expected the non-retryable exception to be rethrown.'); + } catch (RuntimeException $caught) { + self::assertSame($exception, $caught); + } + + self::assertSame(1, $calls); + } + + private function retryableException(): RetryableException + { + return new class ('retryable') extends RuntimeException implements RetryableException { + }; + } + + private function connectionLostException(): ConnectionLost + { + // ConnectionLost is final and its constructor expects a driver + // exception, so build a bare instance for the instanceof check. + return (new ReflectionClass(ConnectionLost::class))->newInstanceWithoutConstructor(); + } +} diff --git a/Tests/Functional/Service/TestableConnection.php b/Tests/Functional/Service/TestableConnection.php new file mode 100644 index 0000000..8286357 --- /dev/null +++ b/Tests/Functional/Service/TestableConnection.php @@ -0,0 +1,37 @@ +dbal = $dbal; + $this->throwableStorage = $throwableStorage; + $this->retryInterval = 0.0; + $this->maxRetries = self::MAX_RETRIES; + } + + public function run(callable $dbalInteraction, ?callable $logContext = null) + { + return $this->withAutoReconnectAndRetry($dbalInteraction, $logContext); + } +} diff --git a/composer.json b/composer.json index 9a2b89b..0aa846f 100644 --- a/composer.json +++ b/composer.json @@ -15,7 +15,7 @@ "neos/flow": "~8.3", "netlogix/jobqueue-polling": "^1.0", "netlogix/jobqueue-pool": "^1.0", - "netlogix/retry": "~1.0", + "netlogix/retry": "^1.1", "php": "~8.2 || ~8.3" }, "require-dev": {