From 91dd7a26e56b9d29682fbced2fd4e242be226d83 Mon Sep 17 00:00:00 2001
From: Albert Chen <albert.cht@gmail.com>
Date: Wed, 28 Oct 2026 17:07:56 +0800
Subject: [PATCH 01/10] fix: retain SQS overflow payloads after ambiguous batch
 failures

The AWS SDK retries SendMessageBatch on connection errors. When SQS
accepted the first attempt but the response was lost, the final
attempt can still report an entry as failed. Treating every failed
entry as never delivered deleted overflow payloads that the accepted
message still referenced. With FIFO deduplication, the later outbox
retry is acknowledged without a replacement message, so the job is lost.

Only clean up entries SQS rejects as sender faults, which every attempt
would have been rejected for. Server faults and entries without the
flag keep their payloads, matching how ambiguous request failures are
already handled. This applies to bulk dispatch and outbox publication.
---
 src/queue/src/SqsQueue.php                    | 20 +++++++--
 .../Queue/Outbox/SqsOutboxPublicationTest.php | 41 ++++++++++++++++++
 tests/Queue/QueueSqsQueueTest.php             | 43 ++++++++++++++++++-
 tests/Queue/SqsBatchLifecycleTest.php         |  2 +-
 4 files changed, 100 insertions(+), 6 deletions(-)

diff --git a/src/queue/src/SqsQueue.php b/src/queue/src/SqsQueue.php
index bddbb4639..ba8cb727b 100644
--- a/src/queue/src/SqsQueue.php
+++ b/src/queue/src/SqsQueue.php
@@ -484,7 +484,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 provably rejected.
             try {
                 $result = $this->sqs->sendMessageBatch([
                     'QueueUrl' => $queueUrl,
@@ -520,7 +520,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->isProvenRejection($rejected)) {
                             $rejectedPaths[] = $overflow[$rejected['Id']]['path'];
                         }
                     }
@@ -612,7 +612,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 own safe cleanup.
                     $result = $this->sqs->sendMessageBatch(['QueueUrl' => $queueUrl, 'Entries' => $chunk]);
                     $chunkIds = array_fill_keys(array_column($chunk, 'Id'), true);
 
@@ -627,7 +627,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->isProvenRejection($rejected)) {
                                     $paths[] = $overflow[$rejected['Id']]['path'];
                                 }
                             }
@@ -672,6 +672,18 @@ protected function batchRejectionException(string $queueUrl, array $chunk, Resul
         );
     }
 
+    /**
+     * Determine if a batch failure proves SQS never accepted the entry.
+     *
+     * SDK retries resend entries SQS may have already accepted, so a server fault
+     * on the final attempt is ambiguous. Sender faults reject the entry's own
+     * content, which every attempt would have been rejected for.
+     */
+    protected function isProvenRejection(array $failure): bool
+    {
+        return ($failure['SenderFault'] ?? false) === true;
+    }
+
     /**
      * Raise queueing-failed events for every outstanding message.
      */
diff --git a/tests/Queue/Outbox/SqsOutboxPublicationTest.php b/tests/Queue/Outbox/SqsOutboxPublicationTest.php
index ecb048348..24296b5ef 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;
@@ -160,6 +162,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/QueueSqsQueueTest.php b/tests/Queue/QueueSqsQueueTest.php
index 1e54c14e2..839673ca8 100644
--- a/tests/Queue/QueueSqsQueueTest.php
+++ b/tests/Queue/QueueSqsQueueTest.php
@@ -1917,7 +1917,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);
@@ -1925,6 +1925,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 8c4343b43..9b249456b 100644
--- a/tests/Queue/SqsBatchLifecycleTest.php
+++ b/tests/Queue/SqsBatchLifecycleTest.php
@@ -130,7 +130,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 = [];

From 333f1c2e922b0aa80d25a94e0d20030a3ca94905 Mon Sep 17 00:00:00 2001
From: Albert Chen <albert.cht@gmail.com>
Date: Wed, 28 Oct 2026 17:08:30 +0800
Subject: [PATCH 02/10] fix: propagate cancellation wrapped by batch
 notification failures

Batch notifications collect listener failures and surface the first one
only after every member has been notified. A failure caused by coroutine
cancellation, such as a cleanup error wrapping CanceledException, was
collected like an ordinary error, so the remaining members kept being
notified after the coroutine was canceled.

Rethrow failures caused by cancellation immediately so the operation
stops instead of continuing normal work.
---
 src/queue/src/Queue.php                    |  6 ++++
 tests/Queue/QueueDatabaseQueueUnitTest.php | 42 ++++++++++++++++++++++
 2 files changed, 48 insertions(+)

diff --git a/src/queue/src/Queue.php b/src/queue/src/Queue.php
index 6e98a2632..c85ebe5fa 100644
--- a/src/queue/src/Queue.php
+++ b/src/queue/src/Queue.php
@@ -15,6 +15,7 @@
 use Hypervel\Contracts\Events\Dispatcher as EventDispatcher;
 use Hypervel\Contracts\Queue\ShouldBeEncrypted;
 use Hypervel\Contracts\Queue\ShouldQueueAfterCommit;
+use Hypervel\Coroutine\Coroutine;
 use Hypervel\Database\DatabaseTransactionsManager;
 use Hypervel\Log\Context\Repository as ContextRepository;
 use Hypervel\Queue\Attributes\Backoff;
@@ -625,6 +626,11 @@ protected function notifyBatchMembers(array $members, Closure ...$notifications)
                 } catch (CanceledException $cancellation) {
                     throw $cancellation;
                 } catch (Throwable $throwable) {
+                    // Collected failures surface only after every member is notified, which is too late for cancellation.
+                    if (Coroutine::causedByCancellation($throwable)) {
+                        throw $throwable;
+                    }
+
                     $exception ??= $throwable;
                 }
             }
diff --git a/tests/Queue/QueueDatabaseQueueUnitTest.php b/tests/Queue/QueueDatabaseQueueUnitTest.php
index da1049d09..213443899 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;
@@ -724,6 +725,47 @@ public function testBulkRaisesExactFailureEventsWhenInsertThrows(): void
         $this->assertSame($exception, $events[3]->exception);
     }
 
+    public function testBulkFailureListenerCancellationIsPropagatedInsteadOfReported(): 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');
+
+        $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(new RuntimeException('Insert failed.'));
+
+        $cancellation = new RuntimeException('Listener cleanup failed.', previous: new CanceledException);
+        $notified = 0;
+        $dispatcher->listen(JobQueueingFailed::class, static function () use (&$notified, $cancellation): void {
+            ++$notified;
+
+            throw $cancellation;
+        });
+        $handler = m::mock(ExceptionHandler::class);
+        $handler->shouldNotReceive('report');
+        $container->instance(ExceptionHandler::class, $handler);
+
+        try {
+            $queue->bulk(['first', 'second'], queue: 'emails');
+            $this->fail('Expected the cancellation to propagate.');
+        } catch (RuntimeException $actual) {
+            $this->assertSame($cancellation, $actual);
+        }
+
+        $this->assertSame(1, $notified);
+    }
+
     public function testBulkCancellationDoesNotRaiseFailureEvents(): void
     {
         $queue = new TestDatabaseQueue(

From cf6d0d2be3d6631ee5ad217cc35f0c95388738c8 Mon Sep 17 00:00:00 2001
From: Albert Chen <albert.cht@gmail.com>
Date: Wed, 28 Oct 2026 17:08:54 +0800
Subject: [PATCH 03/10] fix: keep publishing Redis outbox destinations after
 observer failures

Redis outbox publication stores each destination with its own script
and then notifies observers such as Horizon's JobPushed listeners. An
observer failure was thrown inside the destination loop, so the
remaining destinations on the same connection were never attempted and
their committed rows were released for retry with backoff.

Observers run after the payloads are stored, so continue publishing
the remaining destinations and return the first observer failure with
every accepted row. If a later destination fails to publish, return the
transport failure and report the observer failure separately.
---
 src/queue/src/RedisQueue.php                  | 14 ++++++----
 .../Outbox/RedisOutboxPublicationTest.php     | 27 +++++++++++++++----
 2 files changed, 31 insertions(+), 10 deletions(-)

diff --git a/src/queue/src/RedisQueue.php b/src/queue/src/RedisQueue.php
index 4e02a5416..f5aa16225 100644
--- a/src/queue/src/RedisQueue.php
+++ b/src/queue/src/RedisQueue.php
@@ -13,6 +13,7 @@
 use Hypervel\Contracts\Queue\Queue as QueueContract;
 use Hypervel\Contracts\Redis\Factory as Redis;
 use Hypervel\Database\DatabaseTransactionsManager;
+use Hypervel\ObjectPool\PoolErrorReporter;
 use Hypervel\Queue\Jobs\InspectedJob;
 use Hypervel\Queue\Jobs\RedisJob;
 use Hypervel\Queue\Outbox\OutboxBatchResult;
@@ -465,6 +466,7 @@ protected function enqueueBatch(array $jobs, ?string $queue): void
     public function publishOutboxMessages(array $messages): OutboxBatchResult
     {
         $accepted = [];
+        $observerFailure = null;
 
         if ($messages === []) {
             return new OutboxBatchResult;
@@ -497,19 +499,21 @@ 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 (CanceledException $exception) {
             throw $exception;
         } catch (Throwable $exception) {
+            if ($observerFailure !== null) {
+                PoolErrorReporter::report($observerFailure);
+            }
+
             return new OutboxBatchResult($accepted, failure: $exception);
         }
 
-        return new OutboxBatchResult($accepted);
+        return new OutboxBatchResult($accepted, failure: $observerFailure);
     }
 
     /**
diff --git a/tests/Queue/Outbox/RedisOutboxPublicationTest.php b/tests/Queue/Outbox/RedisOutboxPublicationTest.php
index 0ad0b723e..74160e58b 100644
--- a/tests/Queue/Outbox/RedisOutboxPublicationTest.php
+++ b/tests/Queue/Outbox/RedisOutboxPublicationTest.php
@@ -23,6 +23,7 @@
 use Mockery as m;
 use PHPUnit\Framework\Attributes\DataProvider;
 use RuntimeException;
+use Swoole\Coroutine\CanceledException;
 
 class RedisOutboxPublicationTest extends TestCase
 {
@@ -80,7 +81,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);
@@ -97,16 +98,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);
 
@@ -119,6 +123,19 @@ 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);
+
+        $result = $queue->publishOutboxMessages([$this->message(1), $this->message(2, 'other')]);
+
+        $this->assertSame([1], $result->accepted);
+        $this->assertSame($cancellation, $result->failure);
+    }
+
     /**
      * Create a real driver over an observable transport boundary.
      */

From db9c8a11b39891929aca1a29d62efa2ed6ae39be Mon Sep 17 00:00:00 2001
From: Albert Chen <albert.cht@gmail.com>
Date: Wed, 28 Oct 2026 17:09:30 +0800
Subject: [PATCH 04/10] fix: preserve primary errors when bulk failure
 listeners throw

When a bulk enqueue failed, a JobQueueingFailed listener that threw
replaced the storage or transport error, without chaining it, so callers
and the exception handler lost the root cause. SQS bulk dispatch also
let a JobQueued listener failure replace an explicit SQS rejection.
This affected Redis, database and SQS bulk dispatch and bulk outbox
staging.

Add notifyBatchMembersOfFailure(), which still notifies every member
but reports listener failures through the exception handler instead of
throwing them. Failure paths now rethrow the primary error. Listener
failures that abort an otherwise successful batch are still thrown.
---
 src/queue/src/Concerns/StagesOutboxJobs.php |  4 +-
 src/queue/src/DatabaseQueue.php             |  6 ++-
 src/queue/src/Queue.php                     | 15 ++++++++
 src/queue/src/RedisQueue.php                |  6 ++-
 src/queue/src/SqsQueue.php                  | 21 ++++-------
 tests/Queue/Outbox/OutboxDispatchTest.php   | 28 ++++++++++++++
 tests/Queue/QueueDatabaseQueueUnitTest.php  | 41 +++++++++++++++++++++
 tests/Queue/QueueRedisQueueTest.php         | 37 +++++++++++++++++++
 tests/Queue/SqsBatchLifecycleTest.php       | 20 +++++++---
 9 files changed, 153 insertions(+), 25 deletions(-)

diff --git a/src/queue/src/Concerns/StagesOutboxJobs.php b/src/queue/src/Concerns/StagesOutboxJobs.php
index 35c2406fb..babefa7c3 100644
--- a/src/queue/src/Concerns/StagesOutboxJobs.php
+++ b/src/queue/src/Concerns/StagesOutboxJobs.php
@@ -189,11 +189,11 @@ protected function stageOutboxJobs(array $jobs, mixed $data, ?string $queue): ar
         } catch (CanceledException $exception) {
             throw $exception;
         } catch (Throwable $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 b533b6334..52bc0dd5e 100644
--- a/src/queue/src/DatabaseQueue.php
+++ b/src/queue/src/DatabaseQueue.php
@@ -429,10 +429,12 @@ protected function enqueueBatch(array $jobs, ?string $queue): mixed
         } catch (CanceledException $exception) {
             throw $exception;
         } catch (Throwable $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/Queue.php b/src/queue/src/Queue.php
index c85ebe5fa..1589a4ff4 100644
--- a/src/queue/src/Queue.php
+++ b/src/queue/src/Queue.php
@@ -18,6 +18,7 @@
 use Hypervel\Coroutine\Coroutine;
 use Hypervel\Database\DatabaseTransactionsManager;
 use Hypervel\Log\Context\Repository as ContextRepository;
+use Hypervel\ObjectPool\PoolErrorReporter;
 use Hypervel\Queue\Attributes\Backoff;
 use Hypervel\Queue\Attributes\CountCrashesAsExceptions;
 use Hypervel\Queue\Attributes\Delay;
@@ -639,6 +640,20 @@ 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) {
+            PoolErrorReporter::report($exception);
+        }
+    }
+
     /**
      * Finalize an encoded payload immediately before queueing.
      *
diff --git a/src/queue/src/RedisQueue.php b/src/queue/src/RedisQueue.php
index f5aa16225..d1f1409ef 100644
--- a/src/queue/src/RedisQueue.php
+++ b/src/queue/src/RedisQueue.php
@@ -437,10 +437,12 @@ protected function enqueueBatch(array $jobs, ?string $queue): void
         } catch (CanceledException $exception) {
             throw $exception;
         } catch (Throwable $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/SqsQueue.php b/src/queue/src/SqsQueue.php
index ba8cb727b..81d0e68c7 100644
--- a/src/queue/src/SqsQueue.php
+++ b/src/queue/src/SqsQueue.php
@@ -537,21 +537,20 @@ function (array $success) use ($messages, $queue): void {
                     $this->raiseJobQueuedEvent($queue, $success['MessageId'], $message['job'], $message['payload'], $message['delay']);
                 },
             );
+
+            if ($exception !== null && $notificationFailure !== null) {
+                PoolErrorReporter::report($notificationFailure);
+            }
+
             $exception ??= $notificationFailure;
 
             if ($exception === null) {
                 continue;
             }
 
-            try {
-                $this->raiseBatchFailedEvents($messages, $outstandingMessageIds, $queue, $exception);
-            } catch (CanceledException $cancellation) {
-                throw $cancellation;
-            } catch (Throwable $throwable) {
-                $notificationFailure ??= $throwable;
-            }
+            $this->raiseBatchFailedEvents($messages, $outstandingMessageIds, $queue, $exception);
 
-            throw $notificationFailure ?? $exception;
+            throw $exception;
         }
     }
 
@@ -693,17 +692,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;
-        }
     }
 
     /**
diff --git a/tests/Queue/Outbox/OutboxDispatchTest.php b/tests/Queue/Outbox/OutboxDispatchTest.php
index 402eade6a..0b6b0e7e8 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;
@@ -203,6 +204,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());
+    }
+
     public function testAcceptedBulkIsNotFailedWhenItsQueuedListenerThrows(): void
     {
         $queue = $this->queue();
diff --git a/tests/Queue/QueueDatabaseQueueUnitTest.php b/tests/Queue/QueueDatabaseQueueUnitTest.php
index 213443899..b36d6424a 100644
--- a/tests/Queue/QueueDatabaseQueueUnitTest.php
+++ b/tests/Queue/QueueDatabaseQueueUnitTest.php
@@ -725,6 +725,47 @@ public function testBulkRaisesExactFailureEventsWhenInsertThrows(): void
         $this->assertSame($exception, $events[3]->exception);
     }
 
+    public function testBulkInsertFailureIsNotReplacedByAFailingFailureListener(): 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);
+        $handler->expects('report')->with($listener);
+        $container->instance(ExceptionHandler::class, $handler);
+
+        try {
+            $queue->bulk(['first', 'second'], queue: 'emails');
+            $this->fail('Expected the bulk insert to fail.');
+        } catch (RuntimeException $actual) {
+            $this->assertSame($exception, $actual);
+        }
+
+        $this->assertSame(['first', 'second'], $failed);
+    }
+
     public function testBulkFailureListenerCancellationIsPropagatedInsteadOfReported(): void
     {
         $queue = new TestDatabaseQueue(
diff --git a/tests/Queue/QueueRedisQueueTest.php b/tests/Queue/QueueRedisQueueTest.php
index 42030a671..e0d3c30a5 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;
@@ -160,6 +161,42 @@ public function testBulkUsesOneClusterSafeLuaCall(): void
         $this->assertNull($queue->bulk(['first', 'second']));
     }
 
+    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);
+    }
+
     public function testEmptyBulkDoesNotResolveRedis(): void
     {
         $redis = m::mock(Redis::class);
diff --git a/tests/Queue/SqsBatchLifecycleTest.php b/tests/Queue/SqsBatchLifecycleTest.php
index 9b249456b..201404a16 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;
@@ -96,7 +98,7 @@ public function testPreparationFailureTerminatesEveryOutstandingMember(string $e
 
     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);
@@ -105,12 +107,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'));
@@ -142,12 +147,15 @@ public function testSuccessListenerFailurePreservesAcceptedEntriesAndCleansExpli
         $events->listen(JobQueueingFailed::class, static function (JobQueueingFailed $event) use (&$failed): void {
             $failed[] = $event->job;
         });
+        $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($listener, $caught);
+            $this->fail('The rejection should be rethrown.');
+        } catch (SqsException $caught) {
+            $this->assertSame('InvalidParameterValue', $caught->getAwsErrorCode());
         }
 
         $this->assertSame(array_map('strval', range(1, 9)), $queued);

From 1bf8ddcd73c787e07fe3fde854e643376a53a11d Mon Sep 17 00:00:00 2001
From: Albert Chen <albert.cht@gmail.com>
Date: Wed, 28 Oct 2026 18:09:48 +0800
Subject: [PATCH 05/10] fix: stop outbox delivery when drivers report wrapped
 cancellation

Outbox publication must let coroutine cancellation propagate so the
relay stops without settling the batch. Drivers only rethrew a bare
CanceledException; a failure wrapping cancellation, such as an observer
cleanup error, was returned as an ordinary batch failure. The relay
then settled the accepted rows, released the rest with backoff and
kept running.

Check the publication result in the relay, which covers every driver
including custom ones, and rethrow a failure caused by cancellation.
The Waiter reports it as child cancellation and delivery stops
unsettled, matching how a bare cancellation is handled.
---
 src/queue/src/Outbox/OutboxRelay.php   | 13 ++++++++---
 tests/Queue/Outbox/OutboxRelayTest.php | 32 ++++++++++++++++++++++++++
 2 files changed, 42 insertions(+), 3 deletions(-)

diff --git a/src/queue/src/Outbox/OutboxRelay.php b/src/queue/src/Outbox/OutboxRelay.php
index 9659b1493..13532b23d 100644
--- a/src/queue/src/Outbox/OutboxRelay.php
+++ b/src/queue/src/Outbox/OutboxRelay.php
@@ -9,6 +9,7 @@
 use Hypervel\Contracts\Events\Dispatcher;
 use Hypervel\Contracts\Queue\Factory;
 use Hypervel\Contracts\Queue\PublishesOutboxMessages;
+use Hypervel\Coroutine\Coroutine;
 use Hypervel\Coroutine\Exceptions\ChildCancellationException;
 use Hypervel\Coroutine\Waiter;
 use Hypervel\Queue\Events\OutboxBatchDelivered;
@@ -67,15 +68,21 @@ 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 (CanceledException|ChildCancellationException $exception) {
                     throw $exception;
                 } catch (Throwable $exception) {
-                    return new OutboxBatchResult(failure: $exception);
+                    $result = new OutboxBatchResult(failure: $exception);
+                }
+
+                // Drivers may return a wrapped cancellation as a failure, which must still stop delivery unsettled.
+                if ($result->failure !== null && Coroutine::causedByCancellation($result->failure)) {
+                    throw $result->failure;
                 }
+
+                return $result;
             });
             $duration = (hrtime(true) - $startedAt) / 1_000_000_000;
             $result = $this->validateResult($group, $result);
diff --git a/tests/Queue/Outbox/OutboxRelayTest.php b/tests/Queue/Outbox/OutboxRelayTest.php
index dd935d07a..1a4ac63b0 100644
--- a/tests/Queue/Outbox/OutboxRelayTest.php
+++ b/tests/Queue/Outbox/OutboxRelayTest.php
@@ -278,6 +278,38 @@ public function testIndependentPhaseCancellationReleasesResourcesAndStopsWithout
         $this->assertTrue($released);
     }
 
+    #[DataProvider('wrappedCancellationOutcomes')]
+    public function testWrappedCancellationFromADriverStopsDeliveryWithoutSettlement(bool $returned): void
+    {
+        $repository = m::mock(DatabaseOutboxRepository::class);
+        $repository->shouldNotReceive('settle');
+        $queues = m::mock(Factory::class);
+        $queue = m::mock(Queue::class . ', ' . PublishesOutboxMessages::class);
+        $exceptions = m::mock(ExceptionHandler::class);
+        $exceptions->shouldNotReceive('report');
+        $cancellation = new RuntimeException('Observer cleanup failed.', previous: new CanceledException);
+        $queues->expects('connection')->andReturn($queue);
+        $publication = $queue->expects('publishOutboxMessages');
+        $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)]);
+            $this->fail('Wrapped cancellation must stop delivery.');
+        } catch (ChildCancellationException $exception) {
+            $this->assertSame($cancellation, $exception->getPrevious());
+        }
+    }
+
+    /**
+     * Cover drivers that report cancellation in their result and drivers that throw it.
+     */
+    public static function wrappedCancellationOutcomes(): array
+    {
+        return ['returned' => [true], 'thrown' => [false]];
+    }
+
     /**
      * Build a message whose age and diagnostics are visible to observers.
      */

From f00f100badf67c2cc034c13cc0659a1475645e47 Mon Sep 17 00:00:00 2001
From: Albert Chen <albert.cht@gmail.com>
Date: Wed, 28 Oct 2026 18:10:18 +0800
Subject: [PATCH 06/10] fix: skip queueing failure events for wrapped
 cancellation

Enqueue failure paths skip JobQueueingFailed for a bare
CanceledException, but an exception that wraps cancellation, such as a
cleanup error raised while the coroutine was canceled, still dispatched
the event. In bulk outbox staging this meant every remaining member
received a failure event after a JobQueued listener propagated
cancellation.

Skip the event in raiseJobQueueingFailedEvent() whenever the exception
was caused by cancellation. This covers single and bulk dispatch for
every driver and outbox staging. The exception itself still propagates
unchanged. Document that cancellation does not dispatch the event.
---
 src/docs/queues.md                        |  2 +-
 src/queue/src/Queue.php                   |  5 +++
 tests/Queue/Outbox/OutboxDispatchTest.php | 22 +++++++++++++
 tests/Queue/QueueRedisQueueTest.php       | 40 +++++++++++++++++++++++
 4 files changed, 68 insertions(+), 1 deletion(-)

diff --git a/src/docs/queues.md b/src/docs/queues.md
index 772b87129..11c79464b 100644
--- a/src/docs/queues.md
+++ b/src/docs/queues.md
@@ -4110,7 +4110,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/Queue.php b/src/queue/src/Queue.php
index 1589a4ff4..a495991de 100644
--- a/src/queue/src/Queue.php
+++ b/src/queue/src/Queue.php
@@ -798,6 +798,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/tests/Queue/Outbox/OutboxDispatchTest.php b/tests/Queue/Outbox/OutboxDispatchTest.php
index 0b6b0e7e8..6d475f0a0 100644
--- a/tests/Queue/Outbox/OutboxDispatchTest.php
+++ b/tests/Queue/Outbox/OutboxDispatchTest.php
@@ -34,6 +34,7 @@
 use Pheanstalk\Contract\PheanstalkSubscriberInterface;
 use PHPUnit\Framework\Attributes\DataProvider;
 use RuntimeException;
+use Swoole\Coroutine\CanceledException;
 use UnitEnum;
 
 class OutboxDispatchTest extends TestCase
@@ -231,6 +232,27 @@ public function testBulkFailureIsNotReplacedByAFailingFailureListener(): void
         $this->assertSame(0, DB::table('job_outbox')->count());
     }
 
+    public function testBulkWrappedCancellationStoresNothingAndRaisesNoFailureEvents(): void
+    {
+        $queue = $this->queue();
+        $cancellation = new RuntimeException('Finalization cleanup failed', previous: new CanceledException);
+        $failed = 0;
+        $this->app->make('events')->listen(JobPayloadFinalizing::class, static fn () => throw $cancellation);
+        $this->app->make('events')->listen(JobQueueingFailed::class, function () use (&$failed): void {
+            ++$failed;
+        });
+
+        try {
+            $queue->bulk([(new OutboxDispatchJob)->viaOutbox(), (new OutboxDispatchJob)->viaOutbox()]);
+            $this->fail('Expected the cancellation to propagate.');
+        } catch (RuntimeException $exception) {
+            $this->assertSame($cancellation, $exception);
+        }
+
+        $this->assertSame(0, $failed);
+        $this->assertSame(0, DB::table('job_outbox')->count());
+    }
+
     public function testAcceptedBulkIsNotFailedWhenItsQueuedListenerThrows(): void
     {
         $queue = $this->queue();
diff --git a/tests/Queue/QueueRedisQueueTest.php b/tests/Queue/QueueRedisQueueTest.php
index e0d3c30a5..d5e869879 100644
--- a/tests/Queue/QueueRedisQueueTest.php
+++ b/tests/Queue/QueueRedisQueueTest.php
@@ -35,6 +35,7 @@
 use Mockery as m;
 use PHPUnit\Framework\Attributes\DataProvider;
 use RuntimeException;
+use Swoole\Coroutine\CanceledException;
 use Symfony\Component\Uid\Uuid;
 
 class QueueRedisQueueTest extends TestCase
@@ -197,6 +198,45 @@ public function testBulkRedisFailureIsNotReplacedByAFailingFailureListener(): vo
         $this->assertSame(['first', 'second'], $failed);
     }
 
+    #[DataProvider('enqueueMethods')]
+    public function testWrappedCancellationDoesNotRaiseFailureEvents(string $method): void
+    {
+        $cancellation = new RuntimeException('Redis cleanup failed.', previous: new CanceledException);
+        $connection = $this->mockRedisProxyWithShaCache();
+        $connection->expects('isCluster')->andReturnFalse();
+        $connection->expects('evalWithShaCache')->andThrow($cancellation);
+        $redis = m::mock(Redis::class);
+        $redis->shouldReceive('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');
+        $failed = 0;
+        $events->listen(JobQueueingFailed::class, static function () use (&$failed): void {
+            ++$failed;
+        });
+
+        try {
+            $method === 'bulk' ? $queue->bulk(['first', 'second']) : $queue->push('first');
+            $this->fail('Expected the cancellation to propagate.');
+        } catch (RuntimeException $actual) {
+            $this->assertSame($cancellation, $actual);
+        }
+
+        $this->assertSame(0, $failed);
+    }
+
+    /**
+     * Cover single and set-based enqueue failure paths.
+     */
+    public static function enqueueMethods(): array
+    {
+        return ['push' => ['push'], 'bulk' => ['bulk']];
+    }
+
     public function testEmptyBulkDoesNotResolveRedis(): void
     {
         $redis = m::mock(Redis::class);

From 32ea634861dcf2ef6ac4374e72ac69e6b27ae870 Mon Sep 17 00:00:00 2001
From: Raj Siva-Rajah <5361908+binaryfire@users.noreply.github.com>
Date: Thu, 29 Oct 2026 04:59:48 +0000
Subject: [PATCH 07/10] Preserve cancellation while reporting bulk queue
 failures

---
 src/queue/src/Queue.php                    | 29 ++++++-
 tests/Queue/QueueDatabaseQueueUnitTest.php | 89 ++++++++++------------
 2 files changed, 69 insertions(+), 49 deletions(-)

diff --git a/src/queue/src/Queue.php b/src/queue/src/Queue.php
index 0301e4559..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;
@@ -18,7 +19,6 @@
 use Hypervel\Coroutine\Coroutine;
 use Hypervel\Database\DatabaseTransactionsManager;
 use Hypervel\Log\Context\Repository as ContextRepository;
-use Hypervel\ObjectPool\PoolErrorReporter;
 use Hypervel\Queue\Attributes\Backoff;
 use Hypervel\Queue\Attributes\CountCrashesAsExceptions;
 use Hypervel\Queue\Attributes\Delay;
@@ -655,7 +655,32 @@ protected function notifyBatchMembers(array $members, Closure ...$notifications)
     protected function notifyBatchMembersOfFailure(array $members, Closure $notification): void
     {
         if (($exception = $this->notifyBatchMembers($members, $notification)) !== null) {
-            PoolErrorReporter::report($exception);
+            $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) {
         }
     }
 
diff --git a/tests/Queue/QueueDatabaseQueueUnitTest.php b/tests/Queue/QueueDatabaseQueueUnitTest.php
index d2a52cd08..1d565ef4b 100644
--- a/tests/Queue/QueueDatabaseQueueUnitTest.php
+++ b/tests/Queue/QueueDatabaseQueueUnitTest.php
@@ -25,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;
@@ -44,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;
@@ -727,7 +729,11 @@ public function testBulkRaisesExactFailureEventsWhenInsertThrows(): void
         $this->assertSame($exception, $events[3]->exception);
     }
 
-    public function testBulkInsertFailureIsNotReplacedByAFailingFailureListener(): void
+    #[TestWith(['success'])]
+    #[TestWith(['ordinary'])]
+    #[TestWith(['cancellation'])]
+    #[TestWith(['wrapped'])]
+    public function testBulkInsertFailureListenerErrorReporting(string $reporting): void
     {
         $queue = new TestDatabaseQueue(
             resolver: $resolver = m::mock(ConnectionResolverInterface::class),
@@ -755,58 +761,47 @@ public function testBulkInsertFailureIsNotReplacedByAFailingFailureListener(): v
         });
         $dispatcher->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'], queue: 'emails');
-            $this->fail('Expected the bulk insert to fail.');
-        } catch (RuntimeException $actual) {
-            $this->assertSame($exception, $actual);
+        $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);
         }
-
-        $this->assertSame(['first', 'second'], $failed);
-    }
-
-    public function testBulkFailureListenerCancellationIsPropagatedInsteadOfReported(): 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');
-
-        $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(new RuntimeException('Insert failed.'));
-
-        $cancellation = new RuntimeException('Listener cleanup failed.', previous: new CanceledException);
-        $notified = 0;
-        $dispatcher->listen(JobQueueingFailed::class, static function () use (&$notified, $cancellation): void {
-            ++$notified;
-
-            throw $cancellation;
-        });
-        $handler = m::mock(ExceptionHandler::class);
-        $handler->shouldNotReceive('report');
         $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 {
-            $queue->bulk(['first', 'second'], queue: 'emails');
-            $this->fail('Expected the cancellation to propagate.');
-        } catch (RuntimeException $actual) {
-            $this->assertSame($cancellation, $actual);
+            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(1, $notified);
+        $this->assertSame(['first', 'second'], $failed);
     }
 
     #[TestWith([false])]

From 9434e8f11b685e3576ebef8cae2a8e533ea45d02 Mon Sep 17 00:00:00 2001
From: Raj Siva-Rajah <5361908+binaryfire@users.noreply.github.com>
Date: Thu, 29 Oct 2026 04:59:48 +0000
Subject: [PATCH 08/10] Propagate cancellation from Redis outbox error
 reporting

---
 src/queue/src/RedisQueue.php                  |  3 +--
 .../Outbox/RedisOutboxPublicationTest.php     | 21 +++++++++++++++++++
 2 files changed, 22 insertions(+), 2 deletions(-)

diff --git a/src/queue/src/RedisQueue.php b/src/queue/src/RedisQueue.php
index 74088bc57..09f3b02fc 100644
--- a/src/queue/src/RedisQueue.php
+++ b/src/queue/src/RedisQueue.php
@@ -14,7 +14,6 @@
 use Hypervel\Contracts\Redis\Factory as Redis;
 use Hypervel\Coroutine\Coroutine;
 use Hypervel\Database\DatabaseTransactionsManager;
-use Hypervel\ObjectPool\PoolErrorReporter;
 use Hypervel\Queue\Jobs\InspectedJob;
 use Hypervel\Queue\Jobs\RedisJob;
 use Hypervel\Queue\Outbox\OutboxBatchResult;
@@ -513,7 +512,7 @@ public function publishOutboxMessages(array $messages): OutboxBatchResult
             }
 
             if ($observerFailure !== null) {
-                PoolErrorReporter::report($observerFailure);
+                $this->reportSecondaryFailure($observerFailure);
             }
 
             return new OutboxBatchResult($accepted, failure: $exception);
diff --git a/tests/Queue/Outbox/RedisOutboxPublicationTest.php b/tests/Queue/Outbox/RedisOutboxPublicationTest.php
index 91bc60987..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;
@@ -155,6 +156,26 @@ public function testObserverCancellationStopsPublishingLaterDestinations(): void
         }
     }
 
+    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.
      */

From cb927d386ae5dad422437a830b8160e4ff608ac9 Mon Sep 17 00:00:00 2001
From: Raj Siva-Rajah <5361908+binaryfire@users.noreply.github.com>
Date: Thu, 29 Oct 2026 04:59:48 +0000
Subject: [PATCH 09/10] Preserve SQS reporting cancellation and clarify
 overflow cleanup

---
 src/queue/src/SqsQueue.php            | 25 +++++----
 tests/Queue/SqsBatchLifecycleTest.php | 74 +++++++++++++++++++++++++++
 2 files changed, 88 insertions(+), 11 deletions(-)

diff --git a/src/queue/src/SqsQueue.php b/src/queue/src/SqsQueue.php
index 5ea29b763..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 sender-fault entries 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']]) && $this->isProvenRejection($rejected)) {
+                        if (isset($overflow[$rejected['Id']]) && $this->isSenderFaultRejection($rejected)) {
                             $rejectedPaths[] = $overflow[$rejected['Id']]['path'];
                         }
                     }
@@ -546,7 +546,7 @@ function (array $success) use ($messages, $queue): void {
             );
 
             if ($exception !== null && $notificationFailure !== null) {
-                PoolErrorReporter::report($notificationFailure);
+                $this->reportSecondaryFailure($notificationFailure);
             }
 
             $exception ??= $notificationFailure;
@@ -618,7 +618,7 @@ public function publishOutboxMessages(array $messages): OutboxBatchResult
                         unset($entry);
                     }
 
-                    // SDK failures are ambiguous; only sender-fault 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);
 
@@ -633,7 +633,7 @@ public function publishOutboxMessages(array $messages): OutboxBatchResult
                             $paths = [];
 
                             foreach ($result['Failed'] as $rejected) {
-                                if (isset($chunkIds[$rejected['Id']], $overflow[$rejected['Id']]) && $this->isProvenRejection($rejected)) {
+                                if (isset($chunkIds[$rejected['Id']], $overflow[$rejected['Id']]) && $this->isSenderFaultRejection($rejected)) {
                                     $paths[] = $overflow[$rejected['Id']]['path'];
                                 }
                             }
@@ -681,13 +681,12 @@ protected function batchRejectionException(string $queueUrl, array $chunk, Resul
     }
 
     /**
-     * Determine if a batch failure proves SQS never accepted the entry.
+     * Determine if a failed batch entry's overflow payload may be cleaned up.
      *
-     * SDK retries resend entries SQS may have already accepted, so a server fault
-     * on the final attempt is ambiguous. Sender faults reject the entry's own
-     * content, which every attempt would have been rejected for.
+     * Treat explicit sender-fault entries as rejected. SDK retries can make
+     * server-fault responses ambiguous, so those payloads are retained.
      */
-    protected function isProvenRejection(array $failure): bool
+    protected function isSenderFaultRejection(array $failure): bool
     {
         return ($failure['SenderFault'] ?? false) === true;
     }
@@ -852,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/SqsBatchLifecycleTest.php b/tests/Queue/SqsBatchLifecycleTest.php
index bbb4f2ba6..f55d7f801 100644
--- a/tests/Queue/SqsBatchLifecycleTest.php
+++ b/tests/Queue/SqsBatchLifecycleTest.php
@@ -259,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.
      */

From 1d23cd22c252e4066d588e4cf5a228091e1f1bbf Mon Sep 17 00:00:00 2001
From: Raj Siva-Rajah <5361908+binaryfire@users.noreply.github.com>
Date: Thu, 29 Oct 2026 05:18:15 +0000
Subject: [PATCH 10/10] Stop outbox delivery on returned child cancellation

---
 src/queue/src/Outbox/OutboxRelay.php   | 18 +++++++++++----
 tests/Queue/Outbox/OutboxRelayTest.php | 32 ++++++++++++++++----------
 2 files changed, 33 insertions(+), 17 deletions(-)

diff --git a/src/queue/src/Outbox/OutboxRelay.php b/src/queue/src/Outbox/OutboxRelay.php
index 7a6341e1d..eab599556 100644
--- a/src/queue/src/Outbox/OutboxRelay.php
+++ b/src/queue/src/Outbox/OutboxRelay.php
@@ -71,15 +71,15 @@ public function deliver(string $database, array $messages): bool
                         ? $queue->publishOutboxMessages($group)
                         : $this->publishOutboxMessagesSequentially($queue, $group);
                 } catch (Throwable $exception) {
-                    if ($exception instanceof ChildCancellationException || Coroutine::causedByCancellation($exception)) {
+                    if ($this->shouldStopDelivery($exception)) {
                         throw $exception;
                     }
 
                     $result = new OutboxBatchResult(failure: $exception);
                 }
 
-                // Drivers may return a wrapped cancellation as a failure, which must still stop delivery unsettled.
-                if ($result->failure !== null && Coroutine::causedByCancellation($result->failure)) {
+                // 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;
                 }
 
@@ -95,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;
                     }
 
@@ -115,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.
      *
@@ -196,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/tests/Queue/Outbox/OutboxRelayTest.php b/tests/Queue/Outbox/OutboxRelayTest.php
index 7fb2ce0e7..9cdeef26b 100644
--- a/tests/Queue/Outbox/OutboxRelayTest.php
+++ b/tests/Queue/Outbox/OutboxRelayTest.php
@@ -306,36 +306,44 @@ public function testIndependentPhaseCancellationReleasesResourcesAndStopsDeliver
         $this->assertTrue($released);
     }
 
-    #[DataProvider('wrappedCancellationOutcomes')]
-    public function testWrappedCancellationFromADriverStopsDeliveryWithoutSettlement(bool $returned): void
+    #[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(Queue::class . ', ' . PublishesOutboxMessages::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);
-        $queues->expects('connection')->andReturn($queue);
-        $publication = $queue->expects('publishOutboxMessages');
-        $returned
+        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)]);
-            $this->fail('Wrapped cancellation must stop delivery.');
+            (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, $exception->getPrevious());
+            $this->assertSame($cancellation, $child ? $exception : $exception->getPrevious());
         }
     }
 
     /**
-     * Cover drivers that report cancellation in their result and drivers that throw it.
+     * Cover cancellation thrown by drivers or returned directly and through the sequential fallback.
      */
-    public static function wrappedCancellationOutcomes(): array
+    public static function driverCancellationOutcomes(): array
     {
-        return ['returned' => [true], 'thrown' => [false]];
+        return [
+            'returned wrapped cancellation' => ['returned', false],
+            'thrown wrapped cancellation' => ['thrown', false],
+            'sequential child cancellation' => ['sequential', true],
+            'thrown child cancellation' => ['thrown', true],
+        ];
     }
 
     /**