diff --git a/src/docs/queues.md b/src/docs/queues.md
index 30844080a..11160bd38 100644
--- a/src/docs/queues.md
+++ b/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.
 
diff --git a/src/queue/src/Concerns/StagesOutboxJobs.php b/src/queue/src/Concerns/StagesOutboxJobs.php
index cde9962fe..4dd932ff9 100644
--- a/src/queue/src/Concerns/StagesOutboxJobs.php
+++ b/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);
diff --git a/src/queue/src/DatabaseQueue.php b/src/queue/src/DatabaseQueue.php
index ac189dc2f..89c25988d 100644
--- a/src/queue/src/DatabaseQueue.php
+++ b/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) {
diff --git a/src/queue/src/Outbox/OutboxRelay.php b/src/queue/src/Outbox/OutboxRelay.php
index 034b6807a..eab599556 100644
--- a/src/queue/src/Outbox/OutboxRelay.php
+++ b/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;
                     }
 
diff --git a/src/queue/src/Queue.php b/src/queue/src/Queue.php
index 3c35517ff..d3363bab0 100644
--- a/src/queue/src/Queue.php
+++ b/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');
diff --git a/src/queue/src/RedisQueue.php b/src/queue/src/RedisQueue.php
index 7069558d3..09f3b02fc 100644
--- a/src/queue/src/RedisQueue.php
+++ b/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);
     }
 
     /**
diff --git a/src/queue/src/SqsQueue.php b/src/queue/src/SqsQueue.php
index f84ecfc1d..87f36ef73 100644
--- a/src/queue/src/SqsQueue.php
+++ b/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;
+                    }
                 }
             }
         }
diff --git a/tests/Queue/Outbox/OutboxDispatchTest.php b/tests/Queue/Outbox/OutboxDispatchTest.php
index 52e2d1d9a..e29167467 100644
--- a/tests/Queue/Outbox/OutboxDispatchTest.php
+++ b/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
diff --git a/tests/Queue/Outbox/OutboxRelayTest.php b/tests/Queue/Outbox/OutboxRelayTest.php
index 2b8c0e00e..9cdeef26b 100644
--- a/tests/Queue/Outbox/OutboxRelayTest.php
+++ b/tests/Queue/Outbox/OutboxRelayTest.php
@@ -306,6 +306,46 @@ public function testIndependentPhaseCancellationReleasesResourcesAndStopsDeliver
         $this->assertTrue($released);
     }
 
+    #[DataProvider('driverCancellationOutcomes')]
+    public function testDriverCancellationStopsDeliveryWithoutSettlement(string $outcome, bool $child): void
+    {
+        $repository = m::mock(DatabaseOutboxRepository::class);
+        $repository->shouldNotReceive('settle');
+        $queues = m::mock(Factory::class);
+        $queue = m::mock($outcome === 'sequential' ? Queue::class : Queue::class . ', ' . PublishesOutboxMessages::class);
+        $exceptions = m::mock(ExceptionHandler::class);
+        $exceptions->shouldNotReceive('report');
+        $cancellation = new RuntimeException('Observer cleanup failed.', previous: new CanceledException);
+        if ($child) {
+            $cancellation = new ChildCancellationException('Publication child canceled.', previous: $cancellation);
+        }
+        $queues->expects('connection')->with('redis')->andReturn($queue);
+        $publication = $queue->expects($outcome === 'sequential' ? 'pushRaw' : 'publishOutboxMessages');
+        $outcome === 'returned'
+            ? $publication->andReturn(new OutboxBatchResult([1], failure: $cancellation))
+            : $publication->andThrow($cancellation);
+
+        try {
+            (new OutboxRelay($repository, $queues, new Dispatcher, $exceptions))->deliver('primary', [$this->message(1), $this->message(2, 'other')]);
+            $this->fail('Driver cancellation must stop delivery.');
+        } catch (ChildCancellationException $exception) {
+            $this->assertSame($cancellation, $child ? $exception : $exception->getPrevious());
+        }
+    }
+
+    /**
+     * Cover cancellation thrown by drivers or returned directly and through the sequential fallback.
+     */
+    public static function driverCancellationOutcomes(): array
+    {
+        return [
+            'returned wrapped cancellation' => ['returned', false],
+            'thrown wrapped cancellation' => ['thrown', false],
+            'sequential child cancellation' => ['sequential', true],
+            'thrown child cancellation' => ['thrown', true],
+        ];
+    }
+
     /**
      * Build a message whose age and diagnostics are visible to observers.
      */
diff --git a/tests/Queue/Outbox/RedisOutboxPublicationTest.php b/tests/Queue/Outbox/RedisOutboxPublicationTest.php
index 742986de5..7fd1dbec0 100644
--- a/tests/Queue/Outbox/RedisOutboxPublicationTest.php
+++ b/tests/Queue/Outbox/RedisOutboxPublicationTest.php
@@ -5,6 +5,7 @@
 namespace Hypervel\Tests\Queue\Outbox;
 
 use Hypervel\Container\Container;
+use Hypervel\Contracts\Debug\ExceptionHandler;
 use Hypervel\Contracts\Events\Dispatcher as EventDispatcher;
 use Hypervel\Contracts\Redis\Factory;
 use Hypervel\Events\Dispatcher;
@@ -98,7 +99,7 @@ public function testIncompleteConfirmationRetainsTheEntireDestination(): void
         $this->assertInstanceOf(RuntimeException::class, $result->failure);
     }
 
-    public function testHorizonStampsPublicationAndDrainsAcceptedNotificationsBeforeStopping(): void
+    public function testHorizonStampsPublicationAndKeepsPublishingAfterObserverFailures(): void
     {
         CarbonImmutable::setTestNow('2040-01-01 12:00:00');
         [$queue, $connection, $events] = $this->queue(horizon: true);
@@ -115,16 +116,19 @@ public function testHorizonStampsPublicationAndDrainsAcceptedNotificationsBefore
         });
         $events->listen(JobQueued::class, fn () => $this->fail('Horizon relay must not emit JobQueued.'));
         $stored = [];
-        $connection->expects('evalWithShaCache')->andReturnUsing(function (string $script, array $keys, array $arguments) use (&$stored): int {
-            $stored = [$arguments[1], $arguments[3]];
+        $connection->expects('evalWithShaCache')->twice()->andReturnUsing(function (string $script, array $keys, array $arguments) use (&$stored): int {
+            for ($index = 1; $index < count($arguments); $index += 2) {
+                $stored[] = $arguments[$index];
+            }
 
-            return 2;
+            return intdiv(count($arguments), 2);
         });
 
         $result = $queue->publishOutboxMessages($messages);
 
-        $this->assertSame([1, 2], $result->accepted);
+        $this->assertSame([1, 2, 3], $result->accepted);
         $this->assertSame($failure, $result->failure);
+        $this->assertCount(3, $stored);
         $this->assertSame($stored, $pending);
         $this->assertSame($stored, $pushed);
 
@@ -137,6 +141,41 @@ public function testHorizonStampsPublicationAndDrainsAcceptedNotificationsBefore
         }
     }
 
+    public function testObserverCancellationStopsPublishingLaterDestinations(): void
+    {
+        [$queue, $connection, $events] = $this->queue(horizon: true);
+        $cancellation = new RuntimeException('Observer cleanup failed.', previous: new CanceledException);
+        $events->listen(JobPushed::class, static fn () => throw $cancellation);
+        $connection->expects('evalWithShaCache')->once()->andReturn(1);
+
+        try {
+            $queue->publishOutboxMessages([$this->message(1), $this->message(2, 'other')]);
+            $this->fail('Expected observer cancellation to propagate.');
+        } catch (Throwable $exception) {
+            $this->assertSame($cancellation, $exception);
+        }
+    }
+
+    public function testCancellationDuringObserverFailureReportingPropagates(): void
+    {
+        [$queue, $connection, $events] = $this->queue(horizon: true);
+        $observerFailure = new RuntimeException('Observer failed.');
+        $events->listen(JobPushed::class, static fn () => throw $observerFailure);
+        $connection->expects('evalWithShaCache')->withArgs(fn (string $script, array $keys): bool => $keys[0] === 'queues:first')->andReturn(1);
+        $connection->expects('evalWithShaCache')->withArgs(fn (string $script, array $keys): bool => $keys[0] === 'queues:second')->andThrow(new RuntimeException('Redis unavailable.'));
+        $cancellation = new RuntimeException('Reporting cleanup failed.', previous: new CanceledException);
+        $handler = m::mock(ExceptionHandler::class);
+        $handler->expects('report')->with($observerFailure)->andThrow($cancellation);
+        Container::getInstance()->instance(ExceptionHandler::class, $handler);
+
+        try {
+            $queue->publishOutboxMessages([$this->message(1, 'first'), $this->message(2, 'second')]);
+            $this->fail('Expected reporting cancellation to propagate.');
+        } catch (Throwable $exception) {
+            $this->assertSame($cancellation, $exception);
+        }
+    }
+
     /**
      * Create a real driver over an observable transport boundary.
      */
diff --git a/tests/Queue/Outbox/SqsOutboxPublicationTest.php b/tests/Queue/Outbox/SqsOutboxPublicationTest.php
index 1f18951eb..f0c78bb64 100644
--- a/tests/Queue/Outbox/SqsOutboxPublicationTest.php
+++ b/tests/Queue/Outbox/SqsOutboxPublicationTest.php
@@ -4,6 +4,8 @@
 
 namespace Hypervel\Tests\Queue\Outbox;
 
+use Aws\CommandInterface;
+use Aws\MockHandler;
 use Aws\Result;
 use Aws\Sqs\Exception\SqsException;
 use Aws\Sqs\SqsClient;
@@ -178,6 +180,45 @@ public function testExplicitRejectionCleansOnlyThatAttemptAndRetainsConfirmedEnt
         $this->assertSame($accepted->payload, $store->get($paths[2]));
     }
 
+    public function testServerFaultAfterAnSdkRetryRetainsThePayloadAnEarlierAttemptMayReference(): void
+    {
+        $container = new Container;
+        Container::setInstance($container);
+        $store = new Repository(new WorkerArrayStore);
+        $cache = m::mock(CacheFactory::class);
+        $cache->shouldReceive('store')->with(null)->andReturn($store);
+        $container->instance('cache', $cache);
+        $paths = [];
+        $handler = new MockHandler;
+        $handler->append(
+            function (CommandInterface $command) use (&$paths): SqsException {
+                $paths[] = json_decode($command['Entries'][0]['MessageBody'], true)['@pointer'];
+
+                return new SqsException('Connection reset after SQS accepted the message.', $command, ['connection_error' => true]);
+            },
+            function (CommandInterface $command) use (&$paths): Result {
+                $paths[] = json_decode($command['Entries'][0]['MessageBody'], true)['@pointer'];
+
+                return new Result([
+                    'Successful' => [],
+                    'Failed' => [['Id' => '1', 'Code' => 'InternalError', 'Message' => 'Internal error', 'SenderFault' => false]],
+                ]);
+            },
+        );
+        $client = new SqsClient(['region' => 'us-east-1', 'version' => 'latest', 'credentials' => false, 'handler' => $handler]);
+        $queue = new SqsQueue($client, 'orders.fifo', overflowStorage: ['enabled' => true, 'always' => true]);
+        $queue->setContainer($container);
+        $message = $this->message(1, 'orders.fifo');
+
+        $result = $queue->publishOutboxMessages([$message]);
+
+        $this->assertSame([], $result->accepted);
+        $this->assertInstanceOf(SqsException::class, $result->failure);
+        $this->assertCount(2, $paths);
+        $this->assertSame($paths[0], $paths[1]);
+        $this->assertSame($message->payload, $store->get($paths[0]));
+    }
+
     /**
      * Build a claimed envelope with previously captured FIFO options.
      */
diff --git a/tests/Queue/QueueDatabaseQueueUnitTest.php b/tests/Queue/QueueDatabaseQueueUnitTest.php
index 94b127fcc..1d565ef4b 100644
--- a/tests/Queue/QueueDatabaseQueueUnitTest.php
+++ b/tests/Queue/QueueDatabaseQueueUnitTest.php
@@ -11,6 +11,7 @@
 use Hypervel\Bus\DispatchLockContext;
 use Hypervel\Container\Container;
 use Hypervel\Contracts\Cache\Repository as CacheRepository;
+use Hypervel\Contracts\Debug\ExceptionHandler;
 use Hypervel\Contracts\Events\Dispatcher as DispatcherContract;
 use Hypervel\Contracts\Queue\ShouldQueue;
 use Hypervel\Contracts\Queue\ShouldQueueAfterCommit;
@@ -24,6 +25,7 @@
 use Hypervel\Engine\Channel;
 use Hypervel\Engine\Coroutine as EngineCoroutine;
 use Hypervel\Events\Dispatcher;
+use Hypervel\Filesystem\Filesystem;
 use Hypervel\Queue\Attributes\Backoff;
 use Hypervel\Queue\Attributes\CountCrashesAsExceptions;
 use Hypervel\Queue\Attributes\Delay;
@@ -43,6 +45,7 @@
 use Hypervel\Queue\Queue;
 use Hypervel\Support\CarbonImmutable;
 use Hypervel\Support\Str;
+use Hypervel\Testing\ParallelTesting;
 use Hypervel\Tests\Queue\Fixtures\IntegerQueueName;
 use Hypervel\Tests\Queue\Fixtures\UnitQueueName;
 use Hypervel\Tests\TestCase;
@@ -726,6 +729,81 @@ public function testBulkRaisesExactFailureEventsWhenInsertThrows(): void
         $this->assertSame($exception, $events[3]->exception);
     }
 
+    #[TestWith(['success'])]
+    #[TestWith(['ordinary'])]
+    #[TestWith(['cancellation'])]
+    #[TestWith(['wrapped'])]
+    public function testBulkInsertFailureListenerErrorReporting(string $reporting): void
+    {
+        $queue = new TestDatabaseQueue(
+            resolver: $resolver = m::mock(ConnectionResolverInterface::class),
+            connection: null,
+            table: 'table',
+            default: 'default',
+            currentTime: 1732502704,
+        );
+        Container::setInstance($container = new Container);
+        $dispatcher = new Dispatcher($container);
+        $container->instance('events', $dispatcher);
+        $queue->setContainer($container);
+        $queue->setConnectionName('database');
+
+        $exception = new RuntimeException('Insert failed.');
+        $connection = m::mock(ConnectionInterface::class);
+        $connection->shouldReceive('table')->with('table')->andReturn($query = m::mock(Builder::class));
+        $resolver->shouldReceive('connection')->with(null)->andReturn($connection)->once();
+        $query->shouldReceive('insert')->once()->andThrow($exception);
+
+        $listener = new RuntimeException('Listener failed.');
+        $failed = [];
+        $dispatcher->listen(JobQueueingFailed::class, static function (JobQueueingFailed $event) use (&$failed): void {
+            $failed[] = $event->job;
+        });
+        $dispatcher->listen(JobQueueingFailed::class, static fn () => throw $listener);
+        $handler = m::mock(ExceptionHandler::class);
+        $report = $handler->expects('report')->with($listener);
+        $reportingFailure = match ($reporting) {
+            'ordinary' => new RuntimeException('Reporting failed.'),
+            'cancellation' => new CanceledException,
+            'wrapped' => new RuntimeException('Reporting cleanup failed.', previous: new CanceledException),
+            default => null,
+        };
+        if ($reportingFailure !== null) {
+            $report->andThrow($reportingFailure);
+        }
+        $container->instance(ExceptionHandler::class, $handler);
+        $directory = ParallelTesting::tempDir('QueueDatabaseQueueReportingTest');
+        $files = new Filesystem;
+        $files->deleteDirectory($directory);
+        $files->makeDirectory($directory, recursive: true);
+        $errorLog = $directory . '/errors.log';
+        $previousErrorLog = ini_set('error_log', $errorLog);
+
+        try {
+            try {
+                $queue->bulk(['first', 'second'], queue: 'emails');
+                $this->fail('Expected the bulk insert to fail.');
+            } catch (Throwable $actual) {
+                $this->assertSame(in_array($reporting, ['cancellation', 'wrapped'], true) ? $reportingFailure : $exception, $actual);
+            }
+
+            if ($reporting === 'ordinary') {
+                $contents = $files->get($errorLog);
+                $this->assertStringContainsString('Listener failed.', $contents);
+                $this->assertStringContainsString('Reporting failed.', $contents);
+            } else {
+                $this->assertFileDoesNotExist($errorLog);
+            }
+        } finally {
+            if ($previousErrorLog !== false) {
+                ini_set('error_log', $previousErrorLog);
+            }
+            $files->deleteDirectory($directory);
+        }
+
+        $this->assertSame(['first', 'second'], $failed);
+    }
+
     #[TestWith([false])]
     #[TestWith([true])]
     public function testOutboxSelectionCancellationDoesNotRaiseFailureEvents(bool $cleanupThrows): void
diff --git a/tests/Queue/QueueRedisQueueTest.php b/tests/Queue/QueueRedisQueueTest.php
index 55da0baf9..fdeb94e40 100644
--- a/tests/Queue/QueueRedisQueueTest.php
+++ b/tests/Queue/QueueRedisQueueTest.php
@@ -8,6 +8,7 @@
 use Hypervel\Bus\DispatchLockContext;
 use Hypervel\Container\Container;
 use Hypervel\Contracts\Cache\Repository as CacheRepository;
+use Hypervel\Contracts\Debug\ExceptionHandler;
 use Hypervel\Contracts\Events\Dispatcher;
 use Hypervel\Contracts\Queue\ShouldQueueAfterCommit;
 use Hypervel\Contracts\Redis\Factory as Redis;
@@ -163,9 +164,47 @@ public function testBulkUsesOneClusterSafeLuaCall(): void
         $this->assertNull($queue->bulk(['first', 'second']));
     }
 
-    #[TestWith([false])]
-    #[TestWith([true])]
-    public function testBulkCancellationDoesNotRaiseFailureEvents(bool $cleanupThrows): void
+    public function testBulkRedisFailureIsNotReplacedByAFailingFailureListener(): void
+    {
+        $exception = new RuntimeException('Redis unavailable.');
+        $connection = $this->mockRedisProxyWithShaCache();
+        $connection->expects('isCluster')->andReturnFalse();
+        $connection->expects('evalWithShaCache')->andThrow($exception);
+        $redis = m::mock(Redis::class);
+        $redis->expects('connection')->with('default')->andReturn($connection);
+
+        Container::setInstance($container = new Container);
+        $events = new EventDispatcher($container);
+        $container->instance('events', $events);
+        $queue = new RedisQueue($redis, 'default', 'default');
+        $queue->setContainer($container);
+        $queue->setConnectionName('redis');
+
+        $listener = new RuntimeException('Listener failed.');
+        $failed = [];
+        $events->listen(JobQueueingFailed::class, static function (JobQueueingFailed $event) use (&$failed): void {
+            $failed[] = $event->job;
+        });
+        $events->listen(JobQueueingFailed::class, static fn () => throw $listener);
+        $handler = m::mock(ExceptionHandler::class);
+        $handler->expects('report')->with($listener);
+        $container->instance(ExceptionHandler::class, $handler);
+
+        try {
+            $queue->bulk(['first', 'second']);
+            $this->fail('Expected the Redis failure to be rethrown.');
+        } catch (RuntimeException $actual) {
+            $this->assertSame($exception, $actual);
+        }
+
+        $this->assertSame(['first', 'second'], $failed);
+    }
+
+    #[TestWith(['push', false])]
+    #[TestWith(['push', true])]
+    #[TestWith(['bulk', false])]
+    #[TestWith(['bulk', true])]
+    public function testCancellationDoesNotRaiseFailureEvents(string $method, bool $cleanupThrows): void
     {
         $cancellation = new CanceledException;
         if ($cleanupThrows) {
@@ -175,24 +214,27 @@ public function testBulkCancellationDoesNotRaiseFailureEvents(bool $cleanupThrow
         $connection->expects('isCluster')->andReturnFalse();
         $connection->expects('evalWithShaCache')->andThrow($cancellation);
         $redis = m::mock(Redis::class);
-        $redis->expects('connection')->with('default')->andReturn($connection);
-        $queue = new RedisQueue($redis, 'default', 'default');
-        $events = new EventDispatcher($container = new Container);
+        $redis->shouldReceive('connection')->with('default')->andReturn($connection);
+
+        Container::setInstance($container = new Container);
+        $events = new EventDispatcher($container);
         $container->instance('events', $events);
-        $queue->setContainer($container)->setConnectionName('redis');
-        $failed = [];
-        $events->listen(JobQueueingFailed::class, static function (JobQueueingFailed $event) use (&$failed): void {
-            $failed[] = $event;
+        $queue = new RedisQueue($redis, 'default', 'default');
+        $queue->setContainer($container);
+        $queue->setConnectionName('redis');
+        $failed = 0;
+        $events->listen(JobQueueingFailed::class, static function () use (&$failed): void {
+            ++$failed;
         });
 
         try {
-            $queue->bulk(['first', 'second']);
-            $this->fail('Expected bulk cancellation to propagate.');
-        } catch (Throwable $exception) {
-            $this->assertSame($cancellation, $exception);
+            $method === 'bulk' ? $queue->bulk(['first', 'second']) : $queue->push('first');
+            $this->fail('Expected the cancellation to propagate.');
+        } catch (Throwable $actual) {
+            $this->assertSame($cancellation, $actual);
         }
 
-        $this->assertSame([], $failed);
+        $this->assertSame(0, $failed);
     }
 
     public function testEmptyBulkDoesNotResolveRedis(): void
diff --git a/tests/Queue/QueueSqsQueueTest.php b/tests/Queue/QueueSqsQueueTest.php
index 8fc9ebb5f..23358d91f 100644
--- a/tests/Queue/QueueSqsQueueTest.php
+++ b/tests/Queue/QueueSqsQueueTest.php
@@ -2041,7 +2041,7 @@ public function testBulkCleansOnlyRejectedOverflowPointers(): void
 
         $this->sqs->expects('sendMessageBatch')->andReturn(new Result([
             'Successful' => [['Id' => '0', 'MessageId' => 'accepted-id']],
-            'Failed' => [['Id' => '1', 'Code' => 'Rejected', 'Message' => 'invalid']],
+            'Failed' => [['Id' => '1', 'Code' => 'Rejected', 'Message' => 'invalid', 'SenderFault' => true]],
         ]));
 
         $this->expectException(SqsException::class);
@@ -2049,6 +2049,47 @@ public function testBulkCleansOnlyRejectedOverflowPointers(): void
         $queue->bulk(['a', 'b'], 'data', $this->queueName);
     }
 
+    public function testBulkRetainsOverflowPointersForServerFaultRejections(): void
+    {
+        $payload = json_encode(['uuid' => 'ambiguous'], JSON_THROW_ON_ERROR);
+        $path = SqsQueue::EXTENDED_PAYLOAD_CACHE_PREFIX . 'ambiguous';
+
+        $store = m::mock(CacheRepository::class);
+        $store->expects('put')->with($path, $payload)->andReturnTrue();
+        $store->shouldNotReceive('forget');
+
+        $cache = m::mock(CacheFactory::class);
+        $cache->expects('store')->with('database')->andReturn($store);
+
+        $container = new Container;
+        $container->instance('cache', $cache);
+
+        $queue = $this->getMockBuilder(SqsQueue::class)
+            ->onlyMethods(['getQueue', 'createPayload'])
+            ->setConstructorArgs([
+                $this->sqs,
+                $this->queueName,
+                $this->prefix,
+                '',
+                false,
+                ['enabled' => true, 'always' => true, 'store' => 'database'],
+            ])
+            ->getMock();
+        $queue->setContainer($container);
+        $queue->expects($this->once())->method('getQueue')->willReturn($this->queueUrl);
+        $queue->expects($this->once())->method('createPayload')->willReturn($payload);
+
+        // An SDK retry may have resent an entry SQS already accepted before the final server fault.
+        $this->sqs->expects('sendMessageBatch')->andReturn(new Result([
+            'Successful' => [],
+            'Failed' => [['Id' => '0', 'Code' => 'InternalError', 'Message' => 'failed', 'SenderFault' => false]],
+        ]));
+
+        $this->expectException(SqsException::class);
+
+        $queue->bulk(['a'], 'data', $this->queueName);
+    }
+
     public function testBulkCleansEarlierWritesWhenAWriteFailsBeforeSending(): void
     {
         $firstPayload = json_encode(['uuid' => 'job-1'], JSON_THROW_ON_ERROR);
diff --git a/tests/Queue/SqsBatchLifecycleTest.php b/tests/Queue/SqsBatchLifecycleTest.php
index 3215a76d1..f55d7f801 100644
--- a/tests/Queue/SqsBatchLifecycleTest.php
+++ b/tests/Queue/SqsBatchLifecycleTest.php
@@ -5,6 +5,7 @@
 namespace Hypervel\Tests\Queue;
 
 use Aws\Result;
+use Aws\Sqs\Exception\SqsException;
 use Aws\Sqs\SqsClient;
 use Hypervel\Bus\Dispatcher as BusDispatcher;
 use Hypervel\Bus\DispatchLockContext;
@@ -15,6 +16,7 @@
 use Hypervel\Contracts\Bus\Dispatcher as BusContract;
 use Hypervel\Contracts\Cache\Factory as CacheFactory;
 use Hypervel\Contracts\Cache\Repository as Cache;
+use Hypervel\Contracts\Debug\ExceptionHandler;
 use Hypervel\Contracts\Events\Dispatcher as EventsContract;
 use Hypervel\Contracts\Queue\Queue as QueueContract;
 use Hypervel\Contracts\Queue\ShouldBeUnique;
@@ -132,7 +134,7 @@ public function testPreparationCancellationDoesNotNotifyOutstandingMembers(strin
 
     public function testFailureNotificationsDrainAndRetainThePrimaryAttemptError(): void
     {
-        [$queue, $client, $events] = $this->queue();
+        [$queue, $client, $events, $container] = $this->queue();
         $primary = new RuntimeException('Transport failed.');
         $listener = new RuntimeException('Listener failed.');
         $client->expects('sendMessageBatch')->andThrow($primary);
@@ -141,12 +143,15 @@ public function testFailureNotificationsDrainAndRetainThePrimaryAttemptError():
             $failed[] = $event;
         });
         $events->listen(JobQueueingFailed::class, static fn () => throw $listener);
+        $handler = m::mock(ExceptionHandler::class);
+        $handler->expects('report')->with($listener);
+        $container->instance(ExceptionHandler::class, $handler);
 
         try {
             $queue->bulk(['first', 'second']);
-            $this->fail('The failure listener should throw.');
+            $this->fail('The primary attempt error should be rethrown.');
         } catch (RuntimeException $caught) {
-            $this->assertSame($listener, $caught);
+            $this->assertSame($primary, $caught);
         }
 
         $this->assertSame(['first', 'second'], array_column($failed, 'job'));
@@ -213,7 +218,7 @@ public function testSuccessListenerFailurePreservesAcceptedEntriesAndCleansExpli
 
             return new Result([
                 'Successful' => array_map(static fn (int $id): array => ['Id' => (string) $id, 'MessageId' => 'accepted-' . $id], range(0, 8)),
-                'Failed' => [['Id' => '9', 'Code' => 'InvalidParameterValue', 'Message' => 'Rejected']],
+                'Failed' => [['Id' => '9', 'Code' => 'InvalidParameterValue', 'Message' => 'Rejected', 'SenderFault' => true]],
             ]);
         });
         $queued = $failed = [];
@@ -229,12 +234,20 @@ public function testSuccessListenerFailurePreservesAcceptedEntriesAndCleansExpli
                 throw $cancellation;
             }
         });
+        $handler = m::mock(ExceptionHandler::class);
+        $handler->expects('report')->with($listener);
+        $container->instance(ExceptionHandler::class, $handler);
 
         try {
             $queue->bulk(array_map('strval', range(1, 11)));
-            $this->fail('The monitoring failure should be rethrown.');
-        } catch (RuntimeException $caught) {
-            $this->assertSame($failureListenerCanceled ? $cancellation : $listener, $caught);
+            $this->fail('The rejection should be rethrown.');
+        } catch (Throwable $caught) {
+            if ($failureListenerCanceled) {
+                $this->assertSame($cancellation, $caught);
+            } else {
+                $this->assertInstanceOf(SqsException::class, $caught);
+                $this->assertSame('InvalidParameterValue', $caught->getAwsErrorCode());
+            }
         }
 
         $this->assertSame(array_map('strval', range(1, 9)), $queued);
@@ -246,6 +259,80 @@ public function testSuccessListenerFailurePreservesAcceptedEntriesAndCleansExpli
         }
     }
 
+    #[TestWith([false, false])]
+    #[TestWith([true, false])]
+    #[TestWith([false, true])]
+    #[TestWith([true, true])]
+    public function testCancellationDuringReportingDoesNotNotifyOutstandingMembers(bool $cleanupThrows, bool $overflowCleanup): void
+    {
+        [$queue, $client, $events, $container] = $this->queue(['enabled' => $overflowCleanup, 'always' => true, 'store' => 'overflow']);
+        $client->expects('sendMessageBatch')->andReturn(new Result([
+            'Successful' => [['Id' => '0', 'MessageId' => 'accepted']],
+            'Failed' => [
+                ['Id' => '1', 'Code' => 'InvalidParameterValue', 'Message' => 'Rejected', 'SenderFault' => true],
+                ['Id' => '2', 'Code' => 'InvalidParameterValue', 'Message' => 'Rejected', 'SenderFault' => true],
+            ],
+        ]));
+        $reportedFailure = new RuntimeException($overflowCleanup ? 'Overflow deletion failed.' : 'Queued listener failed.');
+        $deleted = [];
+        if ($overflowCleanup) {
+            $store = m::mock(Cache::class);
+            $store->expects('put')->times(3)->andReturnTrue();
+            $store->expects('forget')->twice()->andReturnUsing(static function (string $path) use (&$deleted, $reportedFailure): bool {
+                $deleted[] = $path;
+                if (count($deleted) === 1) {
+                    throw $reportedFailure;
+                }
+
+                return true;
+            });
+            $cache = m::mock(CacheFactory::class);
+            $cache->expects('store')->with('overflow')->andReturn($store);
+            $container->instance('cache', $cache);
+        }
+        $queued = [];
+        $events->listen(JobQueued::class, static function (JobQueued $event) use (&$queued, $overflowCleanup, $reportedFailure): void {
+            $queued[] = $event->job;
+            if (! $overflowCleanup) {
+                throw $reportedFailure;
+            }
+        });
+        $failed = [];
+        $events->listen(JobQueueingFailed::class, static function (JobQueueingFailed $event) use (&$failed): void {
+            $failed[] = $event;
+        });
+        $gate = $this->armCurrentCoroutineCancellation();
+        $cleanup = new RuntimeException('Reporting cleanup failed.');
+        $handler = m::mock(ExceptionHandler::class);
+        $handler->expects('report')->with($reportedFailure)->andReturnUsing(static function () use ($gate, $cleanup, $cleanupThrows): void {
+            try {
+                $gate->push(true);
+            } finally {
+                if ($cleanupThrows) {
+                    throw $cleanup;
+                }
+            }
+        });
+        $container->instance(ExceptionHandler::class, $handler);
+
+        try {
+            $queue->bulk(['first', 'second', 'third']);
+            $this->fail('Expected reporting cancellation to propagate.');
+        } catch (Throwable $exception) {
+            if ($cleanupThrows) {
+                $this->assertSame($cleanup, $exception);
+                $this->assertInstanceOf(CanceledException::class, $exception->getPrevious());
+            } else {
+                $this->assertInstanceOf(CanceledException::class, $exception);
+            }
+            $this->assertSame([], $failed);
+            $this->assertSame($overflowCleanup ? [] : ['first'], $queued);
+            $this->assertCount($overflowCleanup ? 2 : 0, $deleted);
+        } finally {
+            $gate->close();
+        }
+    }
+
     /**
      * Arm cancellation at a controlled channel handoff.
      */