From 09874b92c0700123f2087974efcb42f56780e119 Mon Sep 17 00:00:00 2001 From: Mark Shust Date: Sun, 1 Nov 2026 08:33:47 -0500 Subject: [PATCH] feat: throw FailedJobNotStoredException when a failed job can't be stored FailedJobHandler now throws a typed FailedJobNotStoredException (a QueueException naming the job class, id, queue and the job's own error, with the storage error as previous) when the failed-job repository throws. failed() and JobFailedEvent still don't run, and callers don't delete the job, so it is failed again once its reservation expires. The worker stops on it: it logs the exception at critical level and rethrows it out of work() (queue:work exits 1), including when a driver throws it from pop(), instead of logging a pop failure and polling again. On the timeout path it is logged at critical level, since it runs inside the alarm handler, and the worker exits as after any timeout. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_017KaHtaJjfe2D38CdEdnmTb --- .../docs/packages/queue-database.md | 2 +- .../docs/packages/queue-redis.md | 2 +- .../docs-markdown/docs/packages/queue-sync.md | 2 +- packages/docs-markdown/docs/packages/queue.md | 24 +- packages/queue-database/src/DatabaseQueue.php | 12 +- .../tests/DatabaseQueueAttemptsTest.php | 88 +++++ packages/queue-redis/src/RedisQueue.php | 6 +- packages/queue-sync/src/SyncQueue.php | 7 +- packages/queue-sync/tests/SyncQueueTest.php | 52 +++ .../FailedJobNotStoredException.php | 37 ++ packages/queue/src/FailedJobHandler.php | 18 +- .../queue/src/FailedJobHandlerInterface.php | 6 +- packages/queue/src/Worker.php | 39 ++- packages/queue/tests/FailedJobHandlerTest.php | 51 ++- .../tests/WorkerFailedJobNotStoredTest.php | 321 ++++++++++++++++++ 15 files changed, 646 insertions(+), 21 deletions(-) create mode 100644 packages/queue/src/Exceptions/FailedJobNotStoredException.php create mode 100644 packages/queue/tests/WorkerFailedJobNotStoredTest.php diff --git a/packages/docs-markdown/docs/packages/queue-database.md b/packages/docs-markdown/docs/packages/queue-database.md index e2cc7531..fbd2fd7f 100644 --- a/packages/docs-markdown/docs/packages/queue-database.md +++ b/packages/docs-markdown/docs/packages/queue-database.md @@ -86,7 +86,7 @@ marko queue:work The `jobs.attempts` column is the authoritative attempt count. Every reservation increments it, so it counts attempts that were *started*: - **A job that throws** is released by the worker. `release()` stores the job the worker hands back in place of the payload it was reserved with, so its attempt count, exception count and retry deadline survive the retry. The job is retried until it has been attempted `maxAttempts` times, then moved to `failed_jobs` and deleted from `jobs`. -- **A job whose worker dies** (fatal error, OOM, `SIGKILL`) is never released. Once its reservation is older than `queue.retry_after`, the next `pop()` reclaims it, and the lost run still counts as an attempt. If the reclaimed job has already used all its attempts, `pop()` hands it to the [failed-job handler](/docs/packages/queue/#when-a-job-fails-for-good) with a `JobAttemptsExhaustedException` ("exceeded max attempts after worker crash or timeout"), so it is stored in `failed_jobs`, its `failed()` hook runs and `JobFailedEvent` is dispatched. Then `pop()` deletes it and moves on to the next job. This happens after the reserving transaction commits, so `failed()` never runs inside it. A job that always crashes its worker can't loop forever. +- **A job whose worker dies** (fatal error, OOM, `SIGKILL`) is never released. Once its reservation is older than `queue.retry_after`, the next `pop()` reclaims it, and the lost run still counts as an attempt. If the reclaimed job has already used all its attempts, `pop()` hands it to the [failed-job handler](/docs/packages/queue/#when-a-job-fails-for-good) with a `JobAttemptsExhaustedException` ("exceeded max attempts after worker crash or timeout"), so it is stored in `failed_jobs`, its `failed()` hook runs and `JobFailedEvent` is dispatched. Then `pop()` deletes it and moves on to the next job. This happens after the reserving transaction commits, so `failed()` never runs inside it. A job that always crashes its worker can't loop forever. If `failed_jobs` can't store the job, `pop()` throws a `FailedJobNotStoredException` and leaves the row reserved, so it is failed again once the reservation expires (see [When a Failed Job Cannot Be Stored](/docs/packages/queue/#when-a-failed-job-cannot-be-stored)). - **A job that releases itself** (calls `release()` inside `handle()`) is stored back by the worker like a job that threw, but its exception count doesn't grow. Its reservation still increments `attempts`. - **A job with a [retry deadline](/docs/packages/queue/#retry-deadlines-exception-limits-and-self-release)** is not limited by attempts, so a polling job that releases itself hundreds of times is never failed at pop time for its attempt count. `pop()` fails it with a `JobRetryDeadlinePassedException` once the deadline has passed instead. diff --git a/packages/docs-markdown/docs/packages/queue-redis.md b/packages/docs-markdown/docs/packages/queue-redis.md index 8d299ecf..d980c8a8 100644 --- a/packages/docs-markdown/docs/packages/queue-redis.md +++ b/packages/docs-markdown/docs/packages/queue-redis.md @@ -107,7 +107,7 @@ A worker that dies mid-job (fatal error, OOM, `SIGKILL`) leaves the job reserved Attempts are counted the way [`marko/queue-database`](/docs/packages/queue-database/#retries-and-attempt-counting) counts them. Every reservation increments the job hash's `attempts`, so a run lost to a crashed worker still counts. `pop()` hands the worker the job with the attempts of the reservations before this one; the worker counts the current run. - **A job that throws** is released by the worker. `release()` takes it off the reserved set, stores the job the worker hands back (its attempts, exception count and retry deadline), and puts it on the ready list or, with a backoff, the delayed set. -- **A job reclaimed after a crash** that has already used all its attempts is failed by `pop()` through the [failed-job handler](/docs/packages/queue/#when-a-job-fails-for-good) with a `JobAttemptsExhaustedException` ("exceeded max attempts after worker crash or timeout"), then deleted. A job past its [retry deadline](/docs/packages/queue/#retry-deadlines-exception-limits-and-self-release) is failed with the deadline error. +- **A job reclaimed after a crash** that has already used all its attempts is failed by `pop()` through the [failed-job handler](/docs/packages/queue/#when-a-job-fails-for-good) with a `JobAttemptsExhaustedException` ("exceeded max attempts after worker crash or timeout"), then deleted. A job past its [retry deadline](/docs/packages/queue/#retry-deadlines-exception-limits-and-self-release) is failed with the deadline error. If the failed-job store can't store the job, `pop()` throws a `FailedJobNotStoredException` and leaves it reserved, so it is failed again once the reservation expires (see [When a Failed Job Cannot Be Stored](/docs/packages/queue/#when-a-failed-job-cannot-be-stored)). The pop-time check uses `RetryPolicy::limitReachedOnReservation()`, the same check `DatabaseQueue` uses. diff --git a/packages/docs-markdown/docs/packages/queue-sync.md b/packages/docs-markdown/docs/packages/queue-sync.md index a9f8fcf2..ae33a702 100644 --- a/packages/docs-markdown/docs/packages/queue-sync.md +++ b/packages/docs-markdown/docs/packages/queue-sync.md @@ -76,7 +76,7 @@ Implements all methods from `QueueInterface`. See [marko/queue](/docs/packages/q | Method | Description | |---|---| | `__construct(JobRunnerInterface $jobRunner, FailedJobHandlerInterface $failedJobHandler, QueueConfig $config)` | Autowired. Every job runs through the runner, and a job that throws goes through the failed-job handler. | -| `push(JobInterface $job, ?string $queue = null): PushResult` | Execute the job immediately through the `JobRunner` and return a pushed `PushResult` with its ID. Container-aware jobs get the container and job envelope first, and release them afterwards. On failure, hands the job to the failed-job handler, then throws `JobFailedException`. | +| `push(JobInterface $job, ?string $queue = null): PushResult` | Execute the job immediately through the `JobRunner` and return a pushed `PushResult` with its ID. Container-aware jobs get the container and job envelope first, and release them afterwards. On failure, hands the job to the failed-job handler, then throws `JobFailedException`, or the handler's `FailedJobNotStoredException` when a bound repository can't store the job. | | `later(int $delay, JobInterface $job, ?string $queue = null): PushResult` | Ignores the delay and executes immediately via `push()`. | | `pop(?string $queue = null, bool $block = false): ?JobInterface` | Always returns `null` --- no jobs are ever queued. | | `size(?string $queue = null): int` | Always returns `0`. | diff --git a/packages/docs-markdown/docs/packages/queue.md b/packages/docs-markdown/docs/packages/queue.md index 38f41e6d..03f002ed 100644 --- a/packages/docs-markdown/docs/packages/queue.md +++ b/packages/docs-markdown/docs/packages/queue.md @@ -397,6 +397,22 @@ A job fails for good when its last attempt throws or times out, when it reaches The caller then deletes the job from its queue. +#### When a Failed Job Cannot Be Stored + +If the failed-job repository throws while storing the job (the `failed_jobs` table is missing, the database is down), the failure isn't recorded anywhere, so `FailedJobHandler` throws a `FailedJobNotStoredException` (a `QueueException`) instead of logging and carrying on. Its message names the job class and id. Its context names the queue, the job's own exception and the storage error, and the storage error is its `getPrevious()`. Then: + +- `failed()` isn't called and `JobFailedEvent` isn't dispatched, so `failed()` never runs twice for one job. +- The caller doesn't delete the job. It stays in its queue (still reserved, for drivers that reserve), so once its reservation expires (`queue.retry_after`) it is popped again and failed again, this time stored if the store works. + +How each caller reacts: + +| Caller | What happens | +|--------|--------------| +| Worker (last attempt, timeout limits, invalid backoff, retry limits) | The worker stops. It [logs](#worker-logs) the exception at `critical` level (`Worker stopping: ...`, with the job, the cause and the suggestion) and throws it out of `work()`, so `queue:work` prints the error and exits with status `1`. It doesn't pull more jobs while the store is broken: every later final failure would fail the same way. Your supervisor restarts the worker; fix the store and the job is failed again on its own. | +| Worker, job timeout | The exception is logged at `critical` level instead of thrown, because the timeout runs inside the alarm signal handler, and the worker exits with status `1` as it does after every timeout. | +| Database and Redis drivers, failing a job at pop time | `pop()` throws the exception and leaves the job reserved. The worker treats it like the case above: it stops instead of logging a [pop failure](#running-the-worker) and polling again. | +| Sync driver | `push()` throws the `FailedJobNotStoredException` instead of `JobFailedException`. `NullFailedJobRepository` never throws, so this only happens when you bind your own repository. | + Implement `FailableJobInterface` to put domain state back when a job gives up, so records don't stay stuck in an in-progress state: ```php @@ -577,7 +593,7 @@ marko queue:work --queue=high,default,low --sleep=1 | `--max-jobs` | Exit after processing this many jobs (default `0`, no limit) | | `--max-time` | Exit after the current job once the worker has run this many seconds (default `0`, no limit) | -If popping a job throws (for example, the broker is unreachable), the worker [logs](#worker-logs) the error, sleeps for `--sleep` seconds and polls again, so a pop failure doesn't stop the worker. With `--once` the error is thrown instead. +If popping a job throws (for example, the broker is unreachable), the worker [logs](#worker-logs) the error, sleeps for `--sleep` seconds and polls again, so a pop failure doesn't stop the worker. With `--once` the error is thrown instead. The one exception is a `FailedJobNotStoredException`, thrown when a driver fails a job at pop time and can't store it: the worker [stops](#when-a-failed-job-cannot-be-stored). #### Worker Logs @@ -588,6 +604,7 @@ The worker and `FailedJobHandler` log through Marko's `LoggerInterface` from [`m | `info` | `Worker stopping: ...`, once per run, with the [stop reason](#stop-reasons-and-exit-codes) | | `warning` | A job sets a timeout but `ext-pcntl` is not loaded, so it can't be enforced | | `error` | A pop failed and the worker will retry, a job exceeded its timeout, a timed-out job could not be released or failed, or a job's `failed()` threw | +| `critical` | `Worker stopping: ...`, when a failed job [could not be stored](#when-a-failed-job-cannot-be-stored) | To send worker logs somewhere else, bind your own `LoggerInterface` implementation or add a handler (see [`marko/log`](/docs/packages/log/)). In tests, pass a `FakeLogger` from [`marko/testing`](/docs/packages/testing/) to `Worker` and `FailedJobHandler` and assert on its entries. @@ -654,6 +671,8 @@ Every way the worker stops goes through one mechanism. `work()` returns a `Worke | `Signal` | SIGTERM or SIGINT arrived (or `stop()` was called) | `0` | | `TimedOut` | A job exceeded its timeout | `1` (`Worker::EXIT_TIMED_OUT`) | +A failed job that [can't be stored](#when-a-failed-job-cannot-be-stored) isn't a stop reason: `work()` throws the `FailedJobNotStoredException`, and `queue:work` exits with status `1`. + The memory, job-count and run-time limits are stop conditions: classes that implement `WorkerStopConditionInterface`. `WorkerOptions::buildStopConditions()` builds `MaxJobsStopCondition`, `MemoryStopCondition` and `MaxTimeStopCondition` from the limits that are on. The worker checks every condition after each job and after each idle pause, with a `WorkerRunState` (when the worker started, the jobs processed so far and the memory in use), and stops with the first reason a condition returns. To stop on something else, implement the interface and pass it through `stopConditions`. Conditions run in order after the built-in ones: @@ -1034,7 +1053,7 @@ use Marko\Queue\JobInterface; public function handle(JobInterface $job, string $queue, Throwable $exception, string $note = ''): void; ``` -Bound to `FailedJobHandler`, which is constructed with the failed-job repository, the `JobEnvelope`, the `JobRunnerInterface`, the `EventDispatcherInterface`, the clock and the `LoggerInterface`. `$note` is appended to the stored exception text. The handler never touches the queue: the caller deletes the job. See [When a Job Fails for Good](#when-a-job-fails-for-good). +Bound to `FailedJobHandler`, which is constructed with the failed-job repository, the `JobEnvelope`, the `JobRunnerInterface`, the `EventDispatcherInterface`, the clock and the `LoggerInterface`. `$note` is appended to the stored exception text. The handler never touches the queue: the caller deletes the job. When the repository can't store the job, `handle()` throws `FailedJobNotStoredException` without calling `failed()` or dispatching the event, and the caller must not delete the job. See [When a Job Fails for Good](#when-a-job-fails-for-good). ### FailableJobInterface @@ -1173,6 +1192,7 @@ Implements `Marko\Core\Event\AsyncObserverDispatcherInterface` and is bound to i |-----------|-------------| | `QueueException` | Base exception for all queue errors --- includes `getContext()` and `getSuggestion()` methods | | `JobFailedException` | Thrown when a job fails during execution | +| `FailedJobNotStoredException` | Thrown by `FailedJobHandler` when the failed-job repository can't store a job that failed for good, with the storage error as `getPrevious()`. See [When a Failed Job Cannot Be Stored](#when-a-failed-job-cannot-be-stored) | | `JobAttemptsExhaustedException` | The reason a job without a retry deadline was not retried once it used its `maxAttempts`. Passed to `failed()` and `JobFailedEvent` for a job that released itself on its last attempt, and when the database driver fails a job whose attempts were all reserved by workers that crashed or timed out | | `JobRetryDeadlinePassedException` | The reason a job was not retried once its retry deadline passed | | `JobExceptionsExhaustedException` | The reason a job was not retried once its exceptions reached `maxExceptions` | diff --git a/packages/queue-database/src/DatabaseQueue.php b/packages/queue-database/src/DatabaseQueue.php index 8b1dd582..e0e46887 100644 --- a/packages/queue-database/src/DatabaseQueue.php +++ b/packages/queue-database/src/DatabaseQueue.php @@ -10,6 +10,7 @@ use Marko\Database\Connection\TransactionInterface; use Marko\Database\Exceptions\LockException; use Marko\Database\Query\QueryBuilderFactoryInterface; +use Marko\Queue\Exceptions\FailedJobNotStoredException; use Marko\Queue\Exceptions\QueueException; use Marko\Queue\Exceptions\SerializationException; use Marko\Queue\FailedJobHandlerInterface; @@ -44,7 +45,9 @@ * crashes its worker cannot be retried forever. A job with a retry deadline * is not limited by attempts, so a polling job that releases itself many * times is not failed here. This happens after the reserving transaction - * commits, so failed() never runs inside it. + * commits, so failed() never runs inside it. When the failed job cannot be + * stored, pop() throws the handler's FailedJobNotStoredException and leaves + * the row reserved, so it is failed again once the reservation expires. * * Reservation: pop() selects the next job with the query builder's * lockForUpdate()->skipLocked() inside a transaction, so concurrent workers @@ -150,7 +153,7 @@ private function table(): string } /** - * @throws LockException|SerializationException + * @throws FailedJobNotStoredException|LockException|SerializationException */ public function pop( ?string $queue = null, @@ -229,9 +232,10 @@ private function reserveNext( /** * Fail a popped job that reached a RetryPolicy limit before this attempt started, then delete it. * - * The row is still reserved, so no other worker picks it up while it is being failed. + * The row is still reserved, so no other worker picks it up while it is being failed. A job that + * cannot be stored is not deleted: the handler throws first. * - * @throws SerializationException + * @throws FailedJobNotStoredException|SerializationException */ private function failExhaustedJob( JobInterface $job, diff --git a/packages/queue-database/tests/DatabaseQueueAttemptsTest.php b/packages/queue-database/tests/DatabaseQueueAttemptsTest.php index fd254103..38906ba8 100644 --- a/packages/queue-database/tests/DatabaseQueueAttemptsTest.php +++ b/packages/queue-database/tests/DatabaseQueueAttemptsTest.php @@ -16,6 +16,7 @@ use Marko\Queue\Database\Tests\Fixtures\TestJob; use Marko\Queue\Database\Tests\Fixtures\TestRetryPolicy; use Marko\Queue\Event\JobFailedEvent; +use Marko\Queue\Exceptions\FailedJobNotStoredException; use Marko\Queue\Exceptions\JobAttemptsExhaustedException; use Marko\Queue\Exceptions\JobRetryDeadlinePassedException; use Marko\Queue\JobEnvelope; @@ -286,3 +287,90 @@ function (): void { expect($queue->pop()?->exceptions)->toBe(1); }); }); + +/** + * A queue (max 1 attempt) whose failed-job handler stores into the connection's failed_jobs table and runs + * failed() with a container that holds the recorder. + * + * @return array{queue: DatabaseQueue, recorder: FailedHookRecorder, events: FakeEventDispatcher} + */ +function notStoredAttemptsQueue( + SqliteConnection $connection, +): array { + $recorder = new FailedHookRecorder($connection); + $container = new Container(); + $container->instance(FailedHookRecorder::class, $recorder); + $events = new FakeEventDispatcher(); + $queue = new DatabaseQueue( + connection: $connection, + jobEnvelope: attemptsEnvelope(), + failedJobHandler: TestFailedJobHandler::create( + new DatabaseFailedJobRepository($connection, DatabaseTimezoneConfig::fromName('UTC')), + attemptsEnvelope(), + new FakeClock(), + $events, + $container, + ), + queryBuilderFactory: new PgSqlQueryBuilderFactory($connection), + clock: new FakeClock(), + databaseTimezoneConfig: DatabaseTimezoneConfig::fromName('UTC'), + retryPolicy: TestRetryPolicy::create(1), + ); + + return ['queue' => $queue, 'recorder' => $recorder, 'events' => $events]; +} + +describe('DatabaseQueue when a crash-exhausted job cannot be stored', function (): void { + it( + 'throws FailedJobNotStoredException from pop() and leaves the job reserved, without calling failed()', + function (): void { + $connection = SqliteConnection::withQueueTables(); + ['queue' => $queue, 'recorder' => $recorder, 'events' => $events] = notStoredAttemptsQueue($connection); + $id = $queue->push(new FailableCrashJob('crashy'), 'emails')->id; + $queue->pop('emails'); + expireReservation($connection, $id); + $connection->execute('ALTER TABLE failed_jobs RENAME TO failed_jobs_gone'); + $thrown = null; + + try { + $queue->pop('emails'); + } catch (FailedJobNotStoredException $e) { + $thrown = $e; + } + + $row = jobsRow($connection, $id); + + expect($thrown)->not->toBeNull() + ->and($thrown->getMessage())->toContain(FailableCrashJob::class . " (id $id)") + ->and($thrown->getContext())->toContain('exceeded max attempts after worker crash or timeout') + ->and($thrown->getPrevious())->not->toBeNull() + ->and($row)->not->toBeNull() + ->and($row['reserved_at'])->not->toBeNull() + ->and($recorder->calls)->toBe([]) + ->and($events->dispatched(JobFailedEvent::class))->toBe([]); + }, + ); + + it( + 'fails the job again, calling failed() once, after the store works and the reservation expires', + function (): void { + $connection = SqliteConnection::withQueueTables(); + ['queue' => $queue, 'recorder' => $recorder] = notStoredAttemptsQueue($connection); + $failedJobs = new DatabaseFailedJobRepository($connection, DatabaseTimezoneConfig::fromName('UTC')); + $id = $queue->push(new FailableCrashJob('crashy'), 'emails')->id; + $queue->pop('emails'); + expireReservation($connection, $id); + $connection->execute('ALTER TABLE failed_jobs RENAME TO failed_jobs_gone'); + + expect(fn () => $queue->pop('emails'))->toThrow(FailedJobNotStoredException::class); + + $connection->execute('ALTER TABLE failed_jobs_gone RENAME TO failed_jobs'); + expireReservation($connection, $id); + + expect($queue->pop('emails'))->toBeNull() + ->and(jobsRow($connection, $id))->toBeNull() + ->and($failedJobs->find($id))->not->toBeNull() + ->and($recorder->calls)->toHaveCount(1); + }, + ); +}); diff --git a/packages/queue-redis/src/RedisQueue.php b/packages/queue-redis/src/RedisQueue.php index 92554244..2e9bfdf6 100644 --- a/packages/queue-redis/src/RedisQueue.php +++ b/packages/queue-redis/src/RedisQueue.php @@ -4,6 +4,7 @@ namespace Marko\Queue\Redis; +use Marko\Queue\Exceptions\FailedJobNotStoredException; use Marko\Queue\Exceptions\SerializationException; use Marko\Queue\FailedJobHandlerInterface; use Marko\Queue\JobEnvelope; @@ -74,7 +75,10 @@ public function later( } /** - * @throws RedisConnectionException|RedisQueueException|SerializationException + * A job failed here that the failed-job repository cannot store stays reserved, and the handler's + * FailedJobNotStoredException is thrown; the job is failed again once the reservation expires. + * + * @throws FailedJobNotStoredException|RedisConnectionException|RedisQueueException|SerializationException */ public function pop( ?string $queue = null, diff --git a/packages/queue-sync/src/SyncQueue.php b/packages/queue-sync/src/SyncQueue.php index 165e458a..3c5b02b8 100644 --- a/packages/queue-sync/src/SyncQueue.php +++ b/packages/queue-sync/src/SyncQueue.php @@ -4,6 +4,7 @@ namespace Marko\Queue\Sync; +use Marko\Queue\Exceptions\FailedJobNotStoredException; use Marko\Queue\Exceptions\JobFailedException; use Marko\Queue\Exceptions\SerializationException; use Marko\Queue\FailedJobHandlerInterface; @@ -30,8 +31,10 @@ public function __construct( * * A job that throws has no further attempts, so it goes through the failed-job handler * (its failed() hook runs and JobFailedEvent is dispatched) before JobFailedException is thrown. + * When a bound failed-job repository cannot store it, the handler's FailedJobNotStoredException is + * thrown instead, and failed() and the event do not run. * - * @throws JobFailedException|RandomException|SerializationException + * @throws FailedJobNotStoredException|JobFailedException|RandomException|SerializationException */ public function push( JobInterface $job, @@ -52,7 +55,7 @@ public function push( } /** - * @throws JobFailedException|RandomException|SerializationException + * @throws FailedJobNotStoredException|JobFailedException|RandomException|SerializationException */ public function later( int $delay, diff --git a/packages/queue-sync/tests/SyncQueueTest.php b/packages/queue-sync/tests/SyncQueueTest.php index d2a511d3..7a8554a2 100644 --- a/packages/queue-sync/tests/SyncQueueTest.php +++ b/packages/queue-sync/tests/SyncQueueTest.php @@ -11,8 +11,10 @@ use Marko\Queue\Event\JobFailedEvent; use Marko\Queue\Event\JobProcessedEvent; use Marko\Queue\Event\JobProcessingEvent; +use Marko\Queue\Exceptions\FailedJobNotStoredException; use Marko\Queue\Exceptions\JobFailedException; use Marko\Queue\FailableJobInterface; +use Marko\Queue\FailedJob; use Marko\Queue\FailedJobHandler; use Marko\Queue\FailedJobHandlerInterface; use Marko\Queue\Job; @@ -390,3 +392,53 @@ public function handle(): void ->and(fn () => $queue->later(60, $job)) ->toThrow(JobFailedException::class); }); + +/** + * Failed-job repository whose store() throws the given error. + */ +class SyncUnwritableFailedJobRepository extends NullFailedJobRepository +{ + public function __construct( + private readonly Throwable $error, + ) {} + + public function store(FailedJob $failedJob): void + { + throw $this->error; + } +} + +it( + 'throws FailedJobNotStoredException instead of JobFailedException when the failed job cannot be stored', + function (): void { + $events = new FakeEventDispatcher(); + $jobEnvelope = createSyncQueueJobEnvelope(); + $storageError = new RuntimeException('failed-job store is down'); + $queue = createSyncQueue( + events: $events, + failedJobHandler: new FailedJobHandler( + new SyncUnwritableFailedJobRepository($storageError), + $jobEnvelope, + new JobRunner(new Container(), $jobEnvelope, $events), + $events, + new FakeClock(), + logger: new FakeLogger(), + ), + ); + $job = new SyncFailableJob(); + $thrown = null; + + try { + $queue->push($job, 'emails'); + } catch (FailedJobNotStoredException $e) { + $thrown = $e; + } + + expect($thrown)->not->toBeNull() + ->and($thrown->getMessage())->toContain(SyncFailableJob::class . " (id $job->id)") + ->and($thrown->getContext())->toContain('sync job failed') + ->and($thrown->getPrevious())->toBe($storageError) + ->and($job->failedWith)->toBe([]) + ->and($events->dispatched(JobFailedEvent::class))->toBe([]); + }, +); diff --git a/packages/queue/src/Exceptions/FailedJobNotStoredException.php b/packages/queue/src/Exceptions/FailedJobNotStoredException.php new file mode 100644 index 00000000..9bb5337c --- /dev/null +++ b/packages/queue/src/Exceptions/FailedJobNotStoredException.php @@ -0,0 +1,37 @@ +getMessage() . '. Storing it then threw ' . $storageError::class . ': ' + . $storageError->getMessage() . '. Its failed() hook was not called and JobFailedEvent was not' + . ' dispatched. The job was not deleted from its queue, so it is failed again once its reservation' + . ' expires (queue.retry_after).', + suggestion: 'Check that the failed-job store is reachable and writable (for marko/queue-database, that' + . ' the failed_jobs table exists and the connection works), then restart the worker.', + previous: $storageError, + ); + } +} diff --git a/packages/queue/src/FailedJobHandler.php b/packages/queue/src/FailedJobHandler.php index 94872c1b..4aa31a01 100644 --- a/packages/queue/src/FailedJobHandler.php +++ b/packages/queue/src/FailedJobHandler.php @@ -7,6 +7,7 @@ use Marko\Core\Event\EventDispatcherInterface; use Marko\Log\Contracts\LoggerInterface; use Marko\Queue\Event\JobFailedEvent; +use Marko\Queue\Exceptions\FailedJobNotStoredException; use Marko\Queue\Exceptions\SerializationException; use Psr\Clock\ClockInterface; use Throwable; @@ -26,7 +27,10 @@ public function __construct( * A failed() hook that throws is logged with the job's class and id and does not stop the * event or the caller: one broken cleanup must not take down a long-running worker. * - * @throws SerializationException + * A job that cannot be stored is a real error, so it throws instead: failed() is not called and the + * event is not dispatched, so failed() never runs twice for a job that is failed again later. + * + * @throws FailedJobNotStoredException|SerializationException */ public function handle( JobInterface $job, @@ -50,7 +54,7 @@ public function handle( * recorded, with a placeholder payload of its class and the serialization error, so the * caller can delete it and keep running. queue:retry refuses that row. * - * @throws SerializationException + * @throws FailedJobNotStoredException|SerializationException */ private function store( JobInterface $job, @@ -71,13 +75,19 @@ private function store( . 'Job class ' . $job::class . ': ' . $serializationError->getMessage(); } - $this->failedJobRepository->store(new FailedJob( + $failedJob = new FailedJob( id: $job->id, queue: $queue, payload: $this->jobEnvelope->wrap($serialized), exception: $exception, failedAt: $this->clock->now(), - )); + ); + + try { + $this->failedJobRepository->store($failedJob); + } catch (Throwable $storageError) { + throw FailedJobNotStoredException::failedJobNotStored($job::class, $job->id, $queue, $e, $storageError); + } } /** diff --git a/packages/queue/src/FailedJobHandlerInterface.php b/packages/queue/src/FailedJobHandlerInterface.php index e374393b..d35b3135 100644 --- a/packages/queue/src/FailedJobHandlerInterface.php +++ b/packages/queue/src/FailedJobHandlerInterface.php @@ -4,6 +4,7 @@ namespace Marko\Queue; +use Marko\Queue\Exceptions\FailedJobNotStoredException; use Marko\Queue\Exceptions\SerializationException; use Throwable; @@ -20,8 +21,11 @@ interface FailedJobHandlerInterface * Store the job as failed, call failed() on a FailableJobInterface job, then dispatch * JobFailedEvent. * + * When the job cannot be stored, it throws without calling failed() or dispatching the event. The + * caller must not delete the job then: it stays in its queue and is failed again later. + * * @param string $note Extra text appended to the stored exception - * @throws SerializationException When the failed job's payload cannot be signed + * @throws FailedJobNotStoredException|SerializationException */ public function handle( JobInterface $job, diff --git a/packages/queue/src/Worker.php b/packages/queue/src/Worker.php index abb0b235..a14e015a 100644 --- a/packages/queue/src/Worker.php +++ b/packages/queue/src/Worker.php @@ -5,6 +5,7 @@ namespace Marko\Queue; use Marko\Log\Contracts\LoggerInterface; +use Marko\Queue\Exceptions\FailedJobNotStoredException; use Marko\Queue\Exceptions\JobTimedOutException; use Marko\Queue\Exceptions\QueueException; use Marko\Queue\Exceptions\SerializationException; @@ -32,8 +33,13 @@ public function __construct( ) {} /** + * A failed job the failed-job repository cannot store ends the run: the worker logs the + * FailedJobNotStoredException at critical level and throws it, so queue:work exits with status 1 instead + * of pulling more jobs while the failed-job store is broken. The job is left in its queue and is failed + * again once its reservation expires. + * * @param list|null $queues Queue names in priority order, or null for the default queue - * @throws QueueException|SerializationException + * @throws FailedJobNotStoredException|QueueException|SerializationException */ public function work( ?array $queues = null, @@ -54,6 +60,9 @@ public function work( while (true) { try { [$job, $queue] = $this->popNextJob($queueNames); + } catch (FailedJobNotStoredException $e) { + // A driver failed a job at pop time and could not store it: the store is broken, so stop + throw $e; } catch (Throwable $e) { // A pop failure (an unreachable broker, a driver error) must not kill a long-running // worker: report it, wait, and poll again. A single --once run reports it loudly instead. @@ -94,6 +103,10 @@ public function work( return $this->stopWith($reason); } } + } catch (FailedJobNotStoredException $e) { + $this->reportFailedJobNotStored($e); + + throw $e; } finally { $this->processControl->stopListening(); } @@ -170,6 +183,9 @@ private function process( * threw. This runs inside the pcntl alarm handler, so a FailableJobInterface job's failed() * runs in signal context too. The process then exits with EXIT_TIMED_OUT so its supervisor * starts a fresh worker. + * + * A FailedJobNotStoredException is logged at critical level rather than thrown: an exception thrown + * from the signal handler would surface inside the interrupted job. The process exits either way. */ private function handleTimeout( JobInterface $job, @@ -189,6 +205,8 @@ private function handleTimeout( try { $this->handleFailedJob($job, $exception, $queue); + } catch (FailedJobNotStoredException $e) { + $this->reportFailedJobNotStored($e); } catch (Throwable $e) { $this->logger->error(sprintf( '[marko/queue] Could not release or fail timed-out job %s: %s: %s', @@ -360,6 +378,21 @@ private function reportPopFailure( )); } + /** + * Log a failed job that could not be stored at critical level, with its context and suggestion, so the + * reason the worker stopped reaches the logs as well as queue:work's output. + */ + private function reportFailedJobNotStored( + FailedJobNotStoredException $e, + ): void { + $this->logger->critical(sprintf( + '[marko/queue] Worker stopping: %s %s %s', + $e->getMessage(), + $e->getContext(), + $e->getSuggestion(), + )); + } + /** * Release a job that threw for another attempt, or fail it for good once RetryPolicy says a limit * is reached (retry deadline, maxExceptions, maxAttempts) or its backoff is invalid. @@ -417,9 +450,9 @@ private function handleReleasedJob( /** * Hand the job to the failed-job handler (store it, call failed(), dispatch JobFailedEvent), - * then delete it from the queue. + * then delete it from the queue. A job the handler could not store is not deleted. * - * @throws SerializationException + * @throws FailedJobNotStoredException|SerializationException */ private function failJob( JobInterface $job, diff --git a/packages/queue/tests/FailedJobHandlerTest.php b/packages/queue/tests/FailedJobHandlerTest.php index 487ca0b1..7d478e02 100644 --- a/packages/queue/tests/FailedJobHandlerTest.php +++ b/packages/queue/tests/FailedJobHandlerTest.php @@ -10,6 +10,8 @@ use Marko\Log\LogLevel; use Marko\Queue\ContainerAwareJobInterface; use Marko\Queue\Event\JobFailedEvent; +use Marko\Queue\Exceptions\FailedJobNotStoredException; +use Marko\Queue\Exceptions\QueueException; use Marko\Queue\FailableJobInterface; use Marko\Queue\FailedJob; use Marko\Queue\FailedJobHandler; @@ -167,8 +169,9 @@ class HandlerCleanupService {} function createLoggingFailedJobHandler( ArrayObject $log = new ArrayObject(), ContainerInterface $container = new Container(), + ?LoggingFailedJobRepository $repository = null, ): array { - $repository = new LoggingFailedJobRepository($log); + $repository ??= new LoggingFailedJobRepository($log); $events = new LoggingEventDispatcher($log); $envelope = createFailedJobHandlerEnvelope(); $logger = new FakeLogger(); @@ -330,4 +333,50 @@ public function failed(Throwable $exception): void ->and($job->failedCalled)->toBeTrue(); }, ); + + it( + 'throws FailedJobNotStoredException naming the job, with the storage error as previous, when storing fails', + function (): void { + $log = new ArrayObject(); + ['handler' => $handler, 'events' => $events, 'logger' => $logger] = createLoggingFailedJobHandler( + $log, + repository: new ThrowingFailedJobRepository(), + ); + $job = new HandlerFailableJob($log); + $job->setId('job-store-broken'); + $storageError = null; + $thrown = null; + + try { + $handler->handle($job, 'emails', new RuntimeException('smtp down')); + } catch (FailedJobNotStoredException $e) { + $thrown = $e; + $storageError = $e->getPrevious(); + } + + expect($thrown)->toBeInstanceOf(QueueException::class) + ->and($thrown->getMessage())->toContain(HandlerFailableJob::class) + ->and($thrown->getMessage())->toContain('job-store-broken') + ->and($thrown->getContext())->toContain('emails') + ->and($thrown->getContext())->toContain('smtp down') + ->and($thrown->getContext())->toContain('failed_jobs table is gone') + ->and($thrown->getSuggestion())->not->toBe('') + ->and($storageError)->toBeInstanceOf(RuntimeException::class) + ->and($storageError?->getMessage())->toBe('failed_jobs table is gone') + ->and($job->failedWith)->toBe([]) + ->and($events->events)->toBe([]) + ->and($logger->entries)->toBe([]); + }, + ); }); + +/** + * Failed-job repository whose store() always throws, like a database that lost its failed_jobs table. + */ +class ThrowingFailedJobRepository extends LoggingFailedJobRepository +{ + public function store(FailedJob $failedJob): void + { + throw new RuntimeException('failed_jobs table is gone'); + } +} diff --git a/packages/queue/tests/WorkerFailedJobNotStoredTest.php b/packages/queue/tests/WorkerFailedJobNotStoredTest.php new file mode 100644 index 00000000..129e92bd --- /dev/null +++ b/packages/queue/tests/WorkerFailedJobNotStoredTest.php @@ -0,0 +1,321 @@ + */ + public array $log = []; + + /** + * @param list $jobs + */ + public function __construct( + private array $jobs, + public ?Throwable $popError = null, + ) {} + + public function push( + JobInterface $job, + ?string $queue = null, + ): PushResult { + return PushResult::pushed('unused'); + } + + public function later( + int $delay, + JobInterface $job, + ?string $queue = null, + ): PushResult { + return PushResult::pushed('unused'); + } + + public function pop( + ?string $queue = null, + bool $block = false, + ): ?JobInterface { + if ($this->popError !== null) { + $this->log[] = 'pop:error'; + + throw $this->popError; + } + + $job = array_shift($this->jobs); + $this->log[] = 'pop:' . ($job?->id ?? 'empty'); + + return $job; + } + + public function size( + ?string $queue = null, + ): int { + return count($this->jobs); + } + + public function clear( + ?string $queue = null, + ): int { + return 0; + } + + public function delete( + string $jobId, + ): bool { + $this->log[] = "delete:$jobId"; + + return true; + } + + public function release( + JobInterface $job, + int $delay = 0, + ): bool { + $this->log[] = "release:$job->id"; + + return true; + } +} + +/** + * Job on its last attempt that runs an optional callback, then throws, and records each failed() call. + */ +class NotStoredFailingJob extends Job implements FailableJobInterface +{ + public int $failedCalls = 0; + + public function __construct( + private readonly ?Closure $during = null, + ) { + $this->maxAttempts = 1; + } + + public function handle(): void + { + if ($this->during !== null) { + ($this->during)(); + } + + throw new RuntimeException('job broke'); + } + + public function failed(Throwable $exception): void + { + $this->failedCalls++; + } +} + +class NotStoredSleepingWorker extends Worker +{ + public int $pauses = 0; + + protected function pause( + int $seconds, + ): void { + $this->pauses++; + $this->stop(); + } +} + +function notStoredJob( + string $id, + ?Closure $during = null, +): NotStoredFailingJob { + $job = new NotStoredFailingJob($during); + $job->setId($id); + + return $job; +} + +/** + * @return array{worker: NotStoredSleepingWorker, logger: FakeLogger} + */ +function createNotStoredWorker( + QueueInterface $queue, + FakeProcessControl $processControl = new FakeProcessControl(), +): array { + $logger = new FakeLogger(); + $envelope = new JobEnvelope( + new EncryptionConfig(new FakeConfigRepository(['encryption.key' => 'worker-not-stored-key'])), + ); + $config = new QueueConfig(new FakeConfigRepository([ + 'queue.queue' => 'default', + 'queue.max_attempts' => 3, + 'queue.retry_after' => 90, + ])); + $worker = new NotStoredSleepingWorker( + $queue, + new FailedJobHandler( + new UnreachableFailedJobRepository(), + $envelope, + new JobRunner(new Container(), $envelope, new FakeEventDispatcher()), + new FakeEventDispatcher(), + new FakeClock(), + $logger, + ), + $config, + new JobRunner(new Container(), $envelope, new FakeEventDispatcher()), + new RetryPolicy($config, new FakeClock()), + new FakeClock(), + $logger, + $processControl, + ); + + return ['worker' => $worker, 'logger' => $logger]; +} + +describe('Worker when a failed job cannot be stored', function (): void { + it('stops with FailedJobNotStoredException instead of pulling more jobs', function (): void { + $job = notStoredJob('broken'); + $queue = new NotStoredRecordingQueue([$job, notStoredJob('next')]); + ['worker' => $worker] = createNotStoredWorker($queue); + + expect(fn () => $worker->work())->toThrow( + FailedJobNotStoredException::class, + 'NotStoredFailingJob (id broken)', + ) + ->and($queue->log)->toBe(['pop:broken']) + ->and($job->failedCalls)->toBe(0); + }); + + it('leaves the job in its queue, neither deleted nor released, so it is failed again later', function (): void { + $queue = new NotStoredRecordingQueue([notStoredJob('broken')]); + ['worker' => $worker] = createNotStoredWorker($queue); + + try { + $worker->work(); + } catch (FailedJobNotStoredException) { + // Expected + } + + expect($queue->log)->not->toContain('delete:broken') + ->and($queue->log)->not->toContain('release:broken'); + }); + + it('logs the failure at critical level with the job, the cause and the suggestion', function (): void { + $queue = new NotStoredRecordingQueue([notStoredJob('broken')]); + ['worker' => $worker, 'logger' => $logger] = createNotStoredWorker($queue); + + try { + $worker->work(); + } catch (FailedJobNotStoredException) { + // Expected + } + + $critical = $logger->entriesForLevel(LogLevel::Critical); + $message = $critical[0]['message'] ?? ''; + + expect($critical)->toHaveCount(1) + ->and($message)->toStartWith('[marko/queue] Worker stopping:') + ->and($message)->toContain('NotStoredFailingJob (id broken)') + ->and($message)->toContain('failed_jobs is unreachable') + ->and($message)->toContain('job broke') + ->and($message)->toContain('restart the worker'); + }); + + it('rethrows FailedJobNotStoredException from pop() instead of logging it and polling again', function (): void { + $notStored = FailedJobNotStoredException::failedJobNotStored( + 'App\Jobs\SendInvoice', + 'job-7', + 'default', + new RuntimeException('attempts used up'), + new RuntimeException('failed_jobs is unreachable'), + ); + $queue = new NotStoredRecordingQueue([], popError: $notStored); + ['worker' => $worker, 'logger' => $logger] = createNotStoredWorker($queue); + + expect(fn () => $worker->work())->toThrow($notStored) + ->and($worker->pauses)->toBe(0) + ->and($queue->log)->toBe(['pop:error']) + ->and($logger->entriesForLevel(LogLevel::Error))->toBe([]) + ->and($logger->entriesForLevel(LogLevel::Critical))->toHaveCount(1); + }); + + it('still logs and polls again after any other pop failure', function (): void { + $queue = new NotStoredRecordingQueue([], popError: new RuntimeException('Broker connection lost')); + ['worker' => $worker, 'logger' => $logger] = createNotStoredWorker($queue); + + $worker->work(); + + expect($worker->pauses)->toBe(1) + ->and($logger->entriesForLevel(LogLevel::Error))->toHaveCount(1) + ->and($logger->entriesForLevel(LogLevel::Critical))->toBe([]); + }); + + it('logs it at critical level and still exits with the timed-out status on a timeout', function (): void { + $process = new FakeProcessControl(); + $queue = new NotStoredRecordingQueue([notStoredJob('slow', fn () => $process->fireTimer())]); + ['worker' => $worker, 'logger' => $logger] = createNotStoredWorker($queue, $process); + + $worker->work(options: new WorkerOptions(timeout: 5)); + + $critical = $logger->entriesForLevel(LogLevel::Critical); + + expect($process->terminatedWith)->toBe(Worker::EXIT_TIMED_OUT) + ->and($critical)->toHaveCount(1) + ->and($critical[0]['message'] ?? '')->toContain('NotStoredFailingJob (id slow)') + ->and($queue->log)->toBe(['pop:slow']); + }); +});