feat: throw when a failed job can't be stored instead of only logging it

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.
 

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.
 

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`. |

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` |

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,

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);
+        },
+    );
+});