From a74e0788105091fea1eba301048f8e0b200d1c20 Mon Sep 17 00:00:00 2001 From: Stephan Schuler Date: Thu, 2 Jul 2026 16:49:23 +0200 Subject: [PATCH 1/4] task: Reformat code --- Classes/Domain/AbstractScheduler.php | 88 +++++++++++++++------------- 1 file changed, 48 insertions(+), 40 deletions(-) diff --git a/Classes/Domain/AbstractScheduler.php b/Classes/Domain/AbstractScheduler.php index 7d1b0e0..5675a56 100644 --- a/Classes/Domain/AbstractScheduler.php +++ b/Classes/Domain/AbstractScheduler.php @@ -51,8 +51,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; } @@ -105,7 +105,7 @@ public function next(string $groupName): ?ScheduledJob */ ->withExponentialBackoff(retryInterval: 0.5, maxRetries: 5) ->onExceptionsOfType(RetryableException::class) - ->task(fn() => $this->dbal + ->task(fn () => $this->dbal ->executeQuery( $claimQuery, [ @@ -148,7 +148,7 @@ public function next(string $groupName): ?ScheduledJob */ ->withExponentialBackoff(retryInterval: 0.5, maxRetries: 5) ->onExceptionsOfType(RetryableException::class) - ->task(fn() => $this->dbal + ->task(fn () => $this->dbal ->executeQuery( $release, [ @@ -166,10 +166,10 @@ public function next(string $groupName): ?ScheduledJob 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,12 +180,13 @@ 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, @@ -196,14 +197,15 @@ public function release(ScheduledJob $job): void ] ); if ($deleteResult->rowCount() === 0) { - $free = /** @lang MySQL */ <<dbal ->executeQuery( $free, @@ -225,13 +227,14 @@ public function fail(ScheduledJob $job, string $reason): void $tableName = ScheduledJob::TABLE_NAME; - $update = /** @lang MySQL */ <<dbal ->executeQuery( $update, @@ -256,17 +259,20 @@ 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, @@ -284,8 +290,8 @@ 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, @@ -295,7 +301,7 @@ public function resetStaleJobs( sql: static::RESET_STALE_JOBS_QUERY, params: [ 'groupName' => $groupName, - 'seconds' => max($minutes === null ? $this->staleJobTimeoutSecs : $minutes * 60, 1), + 'seconds' => max($minutes === null ? $this->staleJobTimeoutSecs : $minutes * 60, 1), ], types: [ 'groupName' => Types::STRING, @@ -315,7 +321,7 @@ protected function scheduleJob(ScheduledJob $job): void */ ->withExponentialBackoff(retryInterval: 0.05, maxRetries: 5) ->onExceptionsOfType(RetryableException::class) - ->task(fn() => $this->dbal + ->task(fn () => $this->dbal ->executeQuery( $statement, [ @@ -350,11 +356,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; } } From 1f875b9948064783e1f4aa4f2f9254678b9149ab Mon Sep 17 00:00:00 2001 From: Stephan Schuler Date: Thu, 2 Jul 2026 17:01:54 +0200 Subject: [PATCH 2/4] feat: Log retryable Exceptions to throwable storage Doctrine RetryableExceptions are either LockWaitTimeoutException or DeadlockException. As the name suggests, they should be safe to retry. This change adds them to the Flow throwable storage anyway, because having those in abundance, at least when it comes to scheduled jobs, might be an indicator of bad database layout or network issues. --- Classes/Domain/AbstractScheduler.php | 72 +++++++++++++++++++++++----- composer.json | 2 +- 2 files changed, 60 insertions(+), 14 deletions(-) diff --git a/Classes/Domain/AbstractScheduler.php b/Classes/Domain/AbstractScheduler.php index 5675a56..5d1cdcd 100644 --- a/Classes/Domain/AbstractScheduler.php +++ b/Classes/Domain/AbstractScheduler.php @@ -9,12 +9,14 @@ use Doctrine\DBAL\Exception\RetryableException; use Doctrine\DBAL\Types\Types; use InvalidArgumentException; +use Neos\Flow\Log\ThrowableStorageInterface; 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; @@ -37,6 +39,11 @@ abstract class AbstractScheduler implements Scheduler */ protected TimeBaseForDueDateCalculation $timeBaseForDueDateCalculation; + /** + * @var ThrowableStorageInterface + */ + protected ThrowableStorageInterface $throwableStorage; + #[Flow\InjectConfiguration(path: 'staleJobTimeout')] protected int $staleJobTimeoutSecs; @@ -56,6 +63,11 @@ public function injectTimeBaseForDueDateCalculation(TimeBaseForDueDateCalculatio $this->timeBaseForDueDateCalculation = $timeBaseForDueDateCalculation; } + public function injectThrowableStorage(ThrowableStorageInterface $throwableStorage): void + { + $this->throwableStorage = $throwableStorage; + } + public function injectSettings(array $settings) { $activeGroupNames = array_filter($settings['groupNames'] ?? []); @@ -105,20 +117,33 @@ public function next(string $groupName): ?ScheduledJob */ ->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, + ->onError( + errorHandler: fn (Throwable $throwable, int $incarnation) => $this->throwableStorage->logThrowable( + throwable: $throwable, + additionalData: [ + 'incarnation' => $incarnation, + 'claim' => $claim, + 'groupName' => $groupName, + 'step' => 'claim', ] - )); + ) + ) + ->task(function () use ($claimQuery, $groupName, $claim) { + return $this->dbal + ->executeQuery( + $claimQuery, + [ + 'now' => $this->timeBaseForDueDateCalculation->getNow(), + 'groupname' => $groupName, + 'claimed' => $claim, + ], + [ + 'now' => Types::DATETIME_IMMUTABLE, + 'groupname' => Types::STRING, + 'claimed' => Types::STRING, + ] + ); + }); $select = static::SELECT_QUERY; @@ -148,6 +173,17 @@ public function next(string $groupName): ?ScheduledJob */ ->withExponentialBackoff(retryInterval: 0.5, maxRetries: 5) ->onExceptionsOfType(RetryableException::class) + ->onError( + errorHandler: fn (Throwable $throwable, int $incarnation) => $this->throwableStorage->logThrowable( + throwable: $throwable, + additionalData: [ + 'incarnation' => $incarnation, + 'claim' => $claim, + 'groupName' => $groupName, + 'step' => 'release', + ] + ) + ) ->task(fn () => $this->dbal ->executeQuery( $release, @@ -321,6 +357,16 @@ protected function scheduleJob(ScheduledJob $job): void */ ->withExponentialBackoff(retryInterval: 0.05, maxRetries: 5) ->onExceptionsOfType(RetryableException::class) + ->onError( + errorHandler: fn (Throwable $throwable, int $incarnation) => $this->throwableStorage->logThrowable( + throwable: $throwable, + additionalData: [ + 'incarnation' => $incarnation, + 'groupName' => $job->getGroupName(), + 'step' => 'schedule', + ] + ) + ) ->task(fn () => $this->dbal ->executeQuery( $statement, 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": { From 485bab9b71954fa53df792fbd8abbec6c75b4f89 Mon Sep 17 00:00:00 2001 From: Stephan Schuler Date: Fri, 3 Jul 2026 14:57:38 +0200 Subject: [PATCH 3/4] fix: Claim jobs by MATERIALIZED CTE instead of locked sub-select 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. --- Classes/Domain/PostgreSQLScheduler.php | 28 +++++++++++++++----- Tests/Functional/RetrievingTest.php | 36 ++++++++++++++++++++++++++ 2 files changed, 58 insertions(+), 6 deletions(-) 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 = <<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'); + } } From 93abc38f01e0fa4ed6db265f6b212c7893c250b4 Mon Sep 17 00:00:00 2001 From: Stephan Schuler Date: Fri, 3 Jul 2026 18:44:57 +0200 Subject: [PATCH 4/4] fix: Unify SQL retry into a single exponential-backoff path The scheduler previously nested two retry mechanisms: an outer Retry (exponential backoff + logging) in AbstractScheduler around the inner Connection::withAutoReconnectAndRetry, which caught ConnectionLost and RetryableException with only a single, non-logged retry attempt. Retry, exponential backoff and throwable logging now live entirely in Connection::withAutoReconnectAndRetry. Both ConnectionLost (reconnect before the next attempt) and RetryableException are retried through the same path. Callers pass an optional $logContext callable to enrich the logged data (step, claim, group name). --- Classes/Domain/AbstractScheduler.php | 228 +++++++----------- Classes/Service/Connection.php | 105 ++++++-- Tests/Functional/RetrievingTest.php | 28 +-- Tests/Functional/SchedulingTest.php | 42 ++-- Tests/Functional/Service/ConnectionTest.php | 211 ++++++++++++++++ .../Functional/Service/TestableConnection.php | 37 +++ 6 files changed, 454 insertions(+), 197 deletions(-) create mode 100644 Tests/Functional/Service/ConnectionTest.php create mode 100644 Tests/Functional/Service/TestableConnection.php diff --git a/Classes/Domain/AbstractScheduler.php b/Classes/Domain/AbstractScheduler.php index 5d1cdcd..42a3a4c 100644 --- a/Classes/Domain/AbstractScheduler.php +++ b/Classes/Domain/AbstractScheduler.php @@ -6,15 +6,12 @@ use DateTimeImmutable; use Doctrine\DBAL\Exception; -use Doctrine\DBAL\Exception\RetryableException; use Doctrine\DBAL\Types\Types; use InvalidArgumentException; -use Neos\Flow\Log\ThrowableStorageInterface; 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; @@ -39,11 +36,6 @@ abstract class AbstractScheduler implements Scheduler */ protected TimeBaseForDueDateCalculation $timeBaseForDueDateCalculation; - /** - * @var ThrowableStorageInterface - */ - protected ThrowableStorageInterface $throwableStorage; - #[Flow\InjectConfiguration(path: 'staleJobTimeout')] protected int $staleJobTimeoutSecs; @@ -63,11 +55,6 @@ public function injectTimeBaseForDueDateCalculation(TimeBaseForDueDateCalculatio $this->timeBaseForDueDateCalculation = $timeBaseForDueDateCalculation; } - public function injectThrowableStorage(ThrowableStorageInterface $throwableStorage): void - { - $this->throwableStorage = $throwableStorage; - } - public function injectSettings(array $settings) { $activeGroupNames = array_filter($settings['groupNames'] ?? []); @@ -111,50 +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) - ->onError( - errorHandler: fn (Throwable $throwable, int $incarnation) => $this->throwableStorage->logThrowable( - throwable: $throwable, - additionalData: [ - 'incarnation' => $incarnation, - 'claim' => $claim, - 'groupName' => $groupName, - 'step' => 'claim', - ] - ) - ) - ->task(function () use ($claimQuery, $groupName, $claim) { - return $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, ] @@ -167,35 +140,23 @@ 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) - ->onError( - errorHandler: fn (Throwable $throwable, int $incarnation) => $this->throwableStorage->logThrowable( - throwable: $throwable, - additionalData: [ - 'incarnation' => $incarnation, - 'claim' => $claim, - 'groupName' => $groupName, - 'step' => 'release', - ] - ) - ) - ->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'], @@ -225,8 +186,8 @@ public function release(ScheduledJob $job): void MySQL; $deleteResult = $this->dbal ->executeQuery( - $delete, - [ + sql: $delete, + params: [ 'groupname' => $job->getGroupName(), 'identifier' => $job->getIdentifier(), 'claimed' => $job->getClaimed(), @@ -244,8 +205,8 @@ public function release(ScheduledJob $job): void MySQL; $this->dbal ->executeQuery( - $free, - [ + sql: $free, + params: [ 'groupname' => $job->getGroupName(), 'identifier' => $job->getIdentifier(), ] @@ -273,13 +234,13 @@ public function fail(ScheduledJob $job, string $reason): void MySQL; $this->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, @@ -311,11 +272,11 @@ public function activity(ScheduledJob $job): void MySQL; $this->dbal ->executeQuery( - $update, - [ + sql: $update, + params: [ 'identifier' => $job->getIdentifier(), ], - [ + types: [ 'identifier' => Types::STRING, ] ); @@ -333,17 +294,18 @@ 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 @@ -351,46 +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) - ->onError( - errorHandler: fn (Throwable $throwable, int $incarnation) => $this->throwableStorage->logThrowable( - throwable: $throwable, - additionalData: [ - 'incarnation' => $incarnation, - 'groupName' => $job->getGroupName(), - 'step' => 'schedule', - ] - ) - ) - ->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. } diff --git a/Classes/Service/Connection.php b/Classes/Service/Connection.php index 07757f5..f032a3c 100644 --- a/Classes/Service/Connection.php +++ b/Classes/Service/Connection.php @@ -10,6 +10,9 @@ use Doctrine\DBAL\Exception\RetryableException; use Doctrine\DBAL\TransactionIsolationLevel; use Doctrine\ORM\EntityManagerInterface; +use Neos\Flow\Log\ThrowableStorageInterface; +use Netlogix\Retry\Retry; +use Throwable; /** * This connection uses the same database credentials as the FLOW @@ -29,6 +32,23 @@ class Connection */ protected $dbal; + /** + * @var ThrowableStorageInterface + */ + protected ThrowableStorageInterface $throwableStorage; + + /** + * @see http://backoffcalculator.com/?attempts=5&rate=1&interval=0.5 + */ + protected float $retryInterval = 0.5; + + protected int $maxRetries = 5; + + public function injectThrowableStorage(ThrowableStorageInterface $throwableStorage): void + { + $this->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 bd0126a..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); } /** 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); + } +}