fix: prevent SQS overflow payload loss and harden queue batch failure handling

src/docs/queues.md

@@ -4128,7 +4128,7 @@ class AppServiceProvider extends ServiceProvider
 
 The `duration` property of the `JobProcessed` event contains the number of milliseconds the job took to run. It is `null` when the worker skips the job because it was already deleted, such as by a `JobProcessing` listener.
 
-Hypervel dispatches a `JobQueueing` event immediately before a job is sent to its queue and a `JobQueued` event after the queue accepts it. If the enqueue attempt throws an exception, a `JobQueueingFailed` event is dispatched with the original exception instead. Jobs deferred until a database transaction commits do not dispatch these events unless the enqueue attempt actually begins.
+Hypervel dispatches a `JobQueueing` event immediately before a job is sent to its queue and a `JobQueued` event after the queue accepts it. If the enqueue attempt throws an exception, a `JobQueueingFailed` event is dispatched with the original exception instead. This event is not dispatched when coroutine cancellation interrupts the attempt, including when the exception was caused by cancellation. Jobs deferred until a database transaction commits do not dispatch these events unless the enqueue attempt actually begins.
 
 The `JobPayloadFinalizing` event runs immediately before `JobQueueing` and may replace its encoded `payload`. It also provides the connection, queue, job, and normalized delay in seconds. Use this event for last-mile payload changes that must reach the queue backend. Listening to both events deliberately runs both listeners for each asynchronous job.
 

src/queue/src/Concerns/StagesOutboxJobs.php

@@ -193,11 +193,11 @@ protected function stageOutboxJobs(array $jobs, mixed $data, ?string $queue): ar
                 throw $exception;
             }
 
-            $failure = $this->notifyBatchMembers($prepared, function (array $member) use ($queue, $exception): void {
+            $this->notifyBatchMembersOfFailure($prepared, function (array $member) use ($queue, $exception): void {
                 $this->raiseJobQueueingFailedEvent($queue, $member['job'], $member['payload'], $member['delay'], $exception);
             });
 
-            throw $failure ?? $exception;
+            throw $exception;
         }
 
         return array_values($direct);

src/queue/src/DatabaseQueue.php

@@ -431,10 +431,12 @@ protected function enqueueBatch(array $jobs, ?string $queue): mixed
                 throw $exception;
             }
 
-            throw $this->notifyBatchMembers(
+            $this->notifyBatchMembersOfFailure(
                 $jobs,
                 fn (array $job) => $this->raiseJobQueueingFailedEvent($queue, $job['job'], $job['payload'], $job['delay'], $exception),
-            ) ?? $exception;
+            );
+
+            throw $exception;
         }
 
         foreach ($jobs as $job) {

src/queue/src/Outbox/OutboxRelay.php

@@ -67,17 +67,23 @@ public function deliver(string $database, array $messages): bool
             $result = $this->inFreshCoroutine(function () use ($connection, $group): OutboxBatchResult {
                 try {
                     $queue = $this->queues->connection($connection);
-
-                    return $queue instanceof PublishesOutboxMessages
+                    $result = $queue instanceof PublishesOutboxMessages
                         ? $queue->publishOutboxMessages($group)
                         : $this->publishOutboxMessagesSequentially($queue, $group);
                 } catch (Throwable $exception) {
-                    if ($exception instanceof ChildCancellationException || Coroutine::causedByCancellation($exception)) {
+                    if ($this->shouldStopDelivery($exception)) {
                         throw $exception;
                     }
 
-                    return new OutboxBatchResult(failure: $exception);
+                    $result = new OutboxBatchResult(failure: $exception);
+                }
+
+                // Drivers and the sequential fallback may return cancellation, including a canceled child's, as a result failure.
+                if ($result->failure !== null && $this->shouldStopDelivery($result->failure)) {
+                    throw $result->failure;
                 }
+
+                return $result;
             });
             $duration = (hrtime(true) - $startedAt) / 1_000_000_000;
             $result = $this->validateResult($group, $result);
@@ -89,7 +95,7 @@ public function deliver(string $database, array $messages): bool
 
                     return null;
                 } catch (Throwable $exception) {
-                    if ($exception instanceof ChildCancellationException || Coroutine::causedByCancellation($exception)) {
+                    if ($this->shouldStopDelivery($exception)) {
                         throw $exception;
                     }
 
@@ -109,6 +115,14 @@ public function deliver(string $database, array $messages): bool
         return $successful;
     }
 
+    /**
+     * Determine whether cancellation requires delivery to stop.
+     */
+    protected function shouldStopDelivery(Throwable $exception): bool
+    {
+        return $exception instanceof ChildCancellationException || Coroutine::causedByCancellation($exception);
+    }
+
     /**
      * Reject driver results that cannot safely identify the rows to settle.
      *
@@ -190,7 +204,7 @@ protected function reportOutcome(
                 try {
                     $this->events->dispatch($event);
                 } catch (Throwable $exception) {
-                    if ($exception instanceof ChildCancellationException || Coroutine::causedByCancellation($exception)) {
+                    if ($this->shouldStopDelivery($exception)) {
                         throw $exception;
                     }
 

src/queue/src/Queue.php

@@ -11,6 +11,7 @@
 use Hypervel\Container\Container as GlobalContainer;
 use Hypervel\Context\CoroutineContext;
 use Hypervel\Contracts\Container\Container;
+use Hypervel\Contracts\Debug\ExceptionHandler;
 use Hypervel\Contracts\Encryption\Encrypter;
 use Hypervel\Contracts\Events\Dispatcher as EventDispatcher;
 use Hypervel\Contracts\Queue\ShouldBeEncrypted;
@@ -644,6 +645,45 @@ protected function notifyBatchMembers(array $members, Closure ...$notifications)
         return $exception;
     }
 
+    /**
+     * Notify every batch member of a failure, reporting listener errors so they cannot replace it.
+     *
+     * @template TMember
+     * @param array<array-key, TMember> $members
+     * @param Closure(TMember): void $notification
+     */
+    protected function notifyBatchMembersOfFailure(array $members, Closure $notification): void
+    {
+        if (($exception = $this->notifyBatchMembers($members, $notification)) !== null) {
+            $this->reportSecondaryFailure($exception);
+        }
+    }
+
+    /**
+     * Report a failure that must not replace the primary error, without swallowing cancellation.
+     */
+    protected function reportSecondaryFailure(Throwable $exception): void
+    {
+        $reportingFailure = null;
+
+        try {
+            if ($this->container->has(ExceptionHandler::class)) {
+                $this->container->make(ExceptionHandler::class)->report($exception);
+
+                return;
+            }
+        } catch (Throwable $reportingFailure) {
+            if ($reportingFailure !== $exception && Coroutine::causedByCancellation($reportingFailure)) {
+                throw $reportingFailure;
+            }
+        }
+
+        try {
+            error_log((string) $exception . ($reportingFailure !== null && $reportingFailure !== $exception ? PHP_EOL . (string) $reportingFailure : ''));
+        } catch (Throwable) {
+        }
+    }
+
     /**
      * Finalize an encoded payload immediately before queueing.
      *
@@ -788,6 +828,11 @@ protected function raiseJobQueueingEvent(?string $queue, object|string $job, str
      */
     protected function raiseJobQueueingFailedEvent(?string $queue, object|string $job, string $payload, DateInterval|DateTimeInterface|int|null $delay, Throwable $exception): void
     {
+        // Cancellation is not a queueing failure; wrapped cancellation skips the event like a bare one.
+        if (Coroutine::causedByCancellation($exception)) {
+            return;
+        }
+
         if ($this->container->bound('events')) {
             /** @var EventDispatcher $events */
             $events = $this->container->make('events');

src/queue/src/RedisQueue.php

@@ -438,10 +438,12 @@ protected function enqueueBatch(array $jobs, ?string $queue): void
                 throw $exception;
             }
 
-            throw $this->notifyBatchMembers(
+            $this->notifyBatchMembersOfFailure(
                 $jobs,
                 fn (array $job) => $this->raiseJobQueueingFailedEvent($queue, $job['job'], $job['payload'], $job['delay'], $exception),
-            ) ?? $exception;
+            );
+
+            throw $exception;
         }
 
         foreach ($jobs as $job) {
@@ -467,6 +469,7 @@ protected function enqueueBatch(array $jobs, ?string $queue): void
     public function publishOutboxMessages(array $messages): OutboxBatchResult
     {
         $accepted = [];
+        $observerFailure = null;
 
         if ($messages === []) {
             return new OutboxBatchResult;
@@ -499,21 +502,23 @@ public function publishOutboxMessages(array $messages): OutboxBatchResult
                     $accepted[] = $message->id;
                 }
 
+                // Observers report after publication, so they must not hold back later destinations.
                 $failure = $this->notifyBatchMembers($prepared, fn (array $member) => $this->handlePayloadPushed($member['payload'], $member['queue']));
-
-                if ($failure !== null) {
-                    throw $failure;
-                }
+                $observerFailure ??= $failure;
             }
         } catch (Throwable $exception) {
             if (Coroutine::causedByCancellation($exception)) {
                 throw $exception;
             }
 
+            if ($observerFailure !== null) {
+                $this->reportSecondaryFailure($observerFailure);
+            }
+
             return new OutboxBatchResult($accepted, failure: $exception);
         }
 
-        return new OutboxBatchResult($accepted);
+        return new OutboxBatchResult($accepted, failure: $observerFailure);
     }
 
     /**

src/queue/src/SqsQueue.php

@@ -489,7 +489,7 @@ protected function sendBatchedMessages(array $messages, ?string $queue): void
             }
 
             // A thrown request is ambiguous for the whole chunk, so retain its pointers.
-            // Only explicit per-entry failures below are provably rejected.
+            // Only sender-fault entries below are treated as rejected.
             try {
                 $result = $this->sqs->sendMessageBatch([
                     'QueueUrl' => $queueUrl,
@@ -527,7 +527,7 @@ protected function sendBatchedMessages(array $messages, ?string $queue): void
                     $rejectedPaths = [];
 
                     foreach ($result['Failed'] as $rejected) {
-                        if (isset($overflow[$rejected['Id']])) {
+                        if (isset($overflow[$rejected['Id']]) && $this->isSenderFaultRejection($rejected)) {
                             $rejectedPaths[] = $overflow[$rejected['Id']]['path'];
                         }
                     }
@@ -544,23 +544,20 @@ function (array $success) use ($messages, $queue): void {
                     $this->raiseJobQueuedEvent($queue, $success['MessageId'], $message['job'], $message['payload'], $message['delay']);
                 },
             );
+
+            if ($exception !== null && $notificationFailure !== null) {
+                $this->reportSecondaryFailure($notificationFailure);
+            }
+
             $exception ??= $notificationFailure;
 
             if ($exception === null) {
                 continue;
             }
 
-            try {
-                $this->raiseBatchFailedEvents($messages, $outstandingMessageIds, $queue, $exception);
-            } catch (Throwable $throwable) {
-                if (Coroutine::causedByCancellation($throwable)) {
-                    throw $throwable;
-                }
-
-                $notificationFailure ??= $throwable;
-            }
+            $this->raiseBatchFailedEvents($messages, $outstandingMessageIds, $queue, $exception);
 
-            throw $notificationFailure ?? $exception;
+            throw $exception;
         }
     }
 
@@ -621,7 +618,7 @@ public function publishOutboxMessages(array $messages): OutboxBatchResult
                         unset($entry);
                     }
 
-                    // SDK failures are ambiguous; only explicit rejects own safe cleanup.
+                    // SDK failures are ambiguous; only sender-fault rejects are cleaned up.
                     $result = $this->sqs->sendMessageBatch(['QueueUrl' => $queueUrl, 'Entries' => $chunk]);
                     $chunkIds = array_fill_keys(array_column($chunk, 'Id'), true);
 
@@ -636,7 +633,7 @@ public function publishOutboxMessages(array $messages): OutboxBatchResult
                             $paths = [];
 
                             foreach ($result['Failed'] as $rejected) {
-                                if (isset($chunkIds[$rejected['Id']], $overflow[$rejected['Id']])) {
+                                if (isset($chunkIds[$rejected['Id']], $overflow[$rejected['Id']]) && $this->isSenderFaultRejection($rejected)) {
                                     $paths[] = $overflow[$rejected['Id']]['path'];
                                 }
                             }
@@ -683,6 +680,17 @@ protected function batchRejectionException(string $queueUrl, array $chunk, Resul
         );
     }
 
+    /**
+     * Determine if a failed batch entry's overflow payload may be cleaned up.
+     *
+     * Treat explicit sender-fault entries as rejected. SDK retries can make
+     * server-fault responses ambiguous, so those payloads are retained.
+     */
+    protected function isSenderFaultRejection(array $failure): bool
+    {
+        return ($failure['SenderFault'] ?? false) === true;
+    }
+
     /**
      * Raise queueing-failed events for every outstanding message.
      */
@@ -692,17 +700,13 @@ protected function raiseBatchFailedEvents(
         ?string $queue,
         Throwable $exception,
     ): void {
-        $notificationFailure = $this->notifyBatchMembers(
+        $this->notifyBatchMembersOfFailure(
             array_keys($outstandingMessageIds),
             function (int $id) use ($messages, $queue, $exception): void {
                 $message = $messages[$id];
                 $this->raiseJobQueueingFailedEvent($queue, $message['job'], $message['payload'], $message['delay'], $exception);
             },
         );
-
-        if ($notificationFailure !== null) {
-            throw $notificationFailure;
-        }
     }
 
     /**
@@ -847,7 +851,11 @@ protected function cleanupOverflowPayloads(CacheRepository $store, array $paths)
                 if ($cancellation === null && Coroutine::causedByCancellation($exception)) {
                     $cancellation = $exception;
                 } else {
-                    PoolErrorReporter::report($exception);
+                    try {
+                        $this->reportSecondaryFailure($exception);
+                    } catch (Throwable $reportingCancellation) {
+                        $cancellation ??= $reportingCancellation;
+                    }
                 }
             }
         }

tests/Queue/Outbox/OutboxDispatchTest.php

@@ -6,6 +6,7 @@
 
 use Aws\Sqs\SqsClient;
 use Hypervel\Bus\Queueable;
+use Hypervel\Contracts\Debug\ExceptionHandler;
 use Hypervel\Contracts\Foundation\Application;
 use Hypervel\Contracts\Queue\Queue as QueueContract;
 use Hypervel\Contracts\Queue\ShouldQueueViaOutbox;
@@ -206,6 +207,33 @@ public function testBulkPreparationFailureStoresNothingAndTerminatesEveryMember(
         $this->assertSame(0, DB::table('job_outbox')->count());
     }
 
+    public function testBulkFailureIsNotReplacedByAFailingFailureListener(): void
+    {
+        $queue = $this->queue();
+        $failure = new RuntimeException('Finalization failed');
+        $listener = new RuntimeException('Listener failed');
+        $failed = [];
+        $this->app->make('events')->listen(JobPayloadFinalizing::class, static fn () => throw $failure);
+        $this->app->make('events')->listen(JobQueueingFailed::class, function (JobQueueingFailed $event) use (&$failed): void {
+            $failed[] = $event->job;
+        });
+        $this->app->make('events')->listen(JobQueueingFailed::class, static fn () => throw $listener);
+        $handler = m::mock(ExceptionHandler::class);
+        $handler->expects('report')->with($listener);
+        $this->app->instance(ExceptionHandler::class, $handler);
+        $jobs = [(new OutboxDispatchJob)->viaOutbox(), (new OutboxDispatchJob)->viaOutbox()];
+
+        try {
+            $queue->bulk($jobs);
+            $this->fail('Expected finalization to fail.');
+        } catch (RuntimeException $exception) {
+            $this->assertSame($failure, $exception);
+        }
+
+        $this->assertSame($jobs, $failed);
+        $this->assertSame(0, DB::table('job_outbox')->count());
+    }
+
     #[TestWith([false])]
     #[TestWith([true])]
     public function testStagingCancellationDoesNotNotifyFailureOrStoreJobs(bool $bulk): void