From ff4bf884f088f83389220d2fdca113f53c77fd2a Mon Sep 17 00:00:00 2001
From: Raj Siva-Rajah <5361908+binaryfire@users.noreply.github.com>
Date: Sat, 12 Sep 2026 15:37:55 +0000
Subject: [PATCH 1/9] Complete deferred callback cleanup for scheduled tasks
Scheduled commands already run in scheduler-owned coroutines, so the console command lifecycle did not drain their deferred callbacks. Run deferred work once after each task and its listeners, using the task outcome and the existing always policy. Keep handled failures coroutine-local without changing protected Laravel method signatures.
Propagate cancellation through scheduled callbacks without converting it into a failed task or running onFailure callbacks. Preserve mutex cleanup and exception precedence. Give paused skip listeners a finite task scope, and emit background-finished notifications only for tasks that were not skipped by overlap or another server.
Cover nested Artisan calls, ordinary and exceptional outcomes, cancellation during tasks and deferred work, background listener ordering, filter and paused callbacks, and both background skip paths. Regression cases reject the previous source. Console and console integration suites, formatting, and full source/type analysis pass.
Investigated during the complete Laravel test cleanup port: https://github.com/laravel/framework/pull/61117. Lifecycle comparison used Laravel 13.x at 01d008c9b5f32cb7c5e50a9a22273113d810b2a2.
---
.../src/Commands/ScheduleRunCommand.php | 106 +++++--
src/console/src/Scheduling/CallbackEvent.php | 4 +
.../Scheduling/ScheduleRunCommandTest.php | 272 +++++++++++++++++-
3 files changed, 361 insertions(+), 21 deletions(-)
diff --git a/src/console/src/Commands/ScheduleRunCommand.php b/src/console/src/Commands/ScheduleRunCommand.php
index d25aea24cc..e469da4972 100644
--- a/src/console/src/Commands/ScheduleRunCommand.php
+++ b/src/console/src/Commands/ScheduleRunCommand.php
@@ -15,6 +15,8 @@
use Hypervel\Console\Scheduling\CallbackEvent;
use Hypervel\Console\Scheduling\Event;
use Hypervel\Console\Scheduling\Schedule;
+use Hypervel\Container\Container;
+use Hypervel\Context\CoroutineContext;
use Hypervel\Contracts\Cache\Repository as Cache;
use Hypervel\Contracts\Debug\ExceptionHandler;
use Hypervel\Contracts\Events\Dispatcher;
@@ -23,10 +25,13 @@
use Hypervel\Log\Context\Repository as ContextRepository;
use Hypervel\Support\CarbonImmutable;
use Hypervel\Support\Collection;
+use Hypervel\Support\Defer\DeferredCallback;
+use Hypervel\Support\Defer\DeferredCallbackCollection;
use Hypervel\Support\Facades\Date;
use Hypervel\Support\InteractsWithTime;
use Hypervel\Support\Sleep;
use RuntimeException;
+use Swoole\Coroutine\CanceledException;
use Symfony\Component\Console\Attribute\AsCommand;
use Throwable;
@@ -35,6 +40,11 @@ class ScheduleRunCommand extends Command
{
use InteractsWithTime;
+ /**
+ * The handled failure state for the current task coroutine.
+ */
+ private const string TASK_FAILED_CONTEXT_KEY = '__console.scheduled_task_failed';
+
/**
* The console command signature.
*/
@@ -248,14 +258,14 @@ protected function repeatEvents(Collection $events): void
continue;
}
- if ($paused && ! $event->runsWhenPaused()) {
- $event->lastChecked = Date::now();
- $this->dispatchTaskSkipped($event);
+ $this->runTaskInCoroutine(function () use ($event, $paused): void {
+ if ($paused && ! $event->runsWhenPaused()) {
+ $event->lastChecked = Date::now();
+ $this->dispatchTaskSkipped($event);
- continue;
- }
+ return;
+ }
- $this->runTaskInCoroutine(function () use ($event): void {
if (! $event->filtersPass($this->hypervel)) {
$this->dispatchTaskSkipped($event);
@@ -285,14 +295,14 @@ protected function runEvents(Collection $events, CarbonInterface $startedAt): vo
continue;
}
- if ($paused && ! $event->runsWhenPaused()) {
- $event->lastChecked = Date::now();
- $this->dispatchTaskSkipped($event);
+ $this->runTaskInCoroutine(function () use ($event, $startedAt, $paused): void {
+ if ($paused && ! $event->runsWhenPaused()) {
+ $event->lastChecked = Date::now();
+ $this->dispatchTaskSkipped($event);
- continue;
- }
+ return;
+ }
- $this->runTaskInCoroutine(function () use ($event, $startedAt): void {
if (! $event->filtersPass($this->hypervel)) {
$this->dispatchTaskSkipped($event);
@@ -312,11 +322,58 @@ protected function runEvents(Collection $events, CarbonInterface $startedAt): vo
protected function runTaskInCoroutine(Closure $callback): void
{
(new Waiter(-1))->wait(
- $callback,
+ fn () => $this->runTask($callback),
copyContext: [ContextRepository::CONTEXT_KEY],
);
}
+ /**
+ * Run a task and its deferred callbacks in the owning coroutine.
+ *
+ * @throws Throwable
+ */
+ private function runTask(Closure $callback): void
+ {
+ $exception = null;
+
+ try {
+ $callback();
+ } catch (Throwable $throwable) {
+ $exception = $throwable;
+ }
+
+ // Nested commands leave deferred work to this boundary. Drain only once,
+ // after task listeners, so callbacks cannot run early or recursively.
+ if (! $exception instanceof CanceledException) {
+ try {
+ $this->invokeDeferredCallbacks(
+ $exception === null && ! CoroutineContext::get(self::TASK_FAILED_CONTEXT_KEY, false)
+ );
+ } catch (CanceledException $cancellation) {
+ $exception = $cancellation;
+ } catch (Throwable $throwable) {
+ $exception ??= $throwable;
+ }
+ }
+
+ if ($exception !== null) {
+ throw $exception;
+ }
+ }
+
+ /**
+ * Invoke deferred callbacks allowed by the task's outcome.
+ */
+ private function invokeDeferredCallbacks(bool $successful): void
+ {
+ $container = Container::getInstance();
+
+ if ($container->resolvedScoped(DeferredCallbackCollection::class)) {
+ $container->make(DeferredCallbackCollection::class)
+ ->invokeWhen(fn (DeferredCallback $callback): bool => $successful || $callback->always);
+ }
+ }
+
/**
* Dispatch a scheduled event in the foreground or background.
*/
@@ -327,13 +384,10 @@ protected function runScheduledEvent(Event $event, CarbonInterface $startedAt):
: $this->runEvent($event);
if ($event->runInBackground) {
- $this->concurrent->fork(function () use ($runEvent, $event): void {
- $runEvent();
-
- if ($this->dispatcher->hasListeners(ScheduledBackgroundTaskFinished::class)) {
- $this->dispatcher->dispatch(new ScheduledBackgroundTaskFinished($event));
- }
- }, [ContextRepository::CONTEXT_KEY]);
+ $this->concurrent->fork(
+ fn () => $this->runTask($runEvent),
+ [ContextRepository::CONTEXT_KEY],
+ );
return;
}
@@ -440,8 +494,13 @@ protected function runEvent(Event $event): void
"Scheduled command [{$command}] failed with exit code [{$exitCode}]."
);
}
+ } catch (CanceledException $e) {
+ throw $e;
} catch (Throwable $e) {
$failed = true;
+ // runEvent reports ordinary failures instead of throwing them to
+ // runTask; keep that outcome local to this task's deferred work.
+ CoroutineContext::set(self::TASK_FAILED_CONTEXT_KEY, true);
if ($this->dispatcher->hasListeners(ScheduledTaskFailed::class)) {
$this->dispatcher->dispatch(new ScheduledTaskFailed($event, $e));
@@ -468,6 +527,13 @@ protected function runEvent(Event $event): void
);
$this->line($finishDescription);
+
+ if ($event->runInBackground
+ && ! $skippedBecauseOverlapping
+ && $this->dispatcher->hasListeners(ScheduledBackgroundTaskFinished::class)
+ ) {
+ $this->dispatcher->dispatch(new ScheduledBackgroundTaskFinished($event));
+ }
}
/**
diff --git a/src/console/src/Scheduling/CallbackEvent.php b/src/console/src/Scheduling/CallbackEvent.php
index 21c6c64445..a14437d61b 100644
--- a/src/console/src/Scheduling/CallbackEvent.php
+++ b/src/console/src/Scheduling/CallbackEvent.php
@@ -11,6 +11,7 @@
use InvalidArgumentException;
use LogicException;
use RuntimeException;
+use Swoole\Coroutine\CanceledException;
use Throwable;
class CallbackEvent extends Event
@@ -126,6 +127,9 @@ protected function execute(Container $container): int
]);
return $result === false ? 1 : 0;
+ } catch (CanceledException $e) {
+ // Cancellation must unwind the mutex without running failure callbacks.
+ throw $e;
} catch (Throwable $e) {
CoroutineContext::set($this->callbackContextKey(), [
'result' => null,
diff --git a/tests/Console/Scheduling/ScheduleRunCommandTest.php b/tests/Console/Scheduling/ScheduleRunCommandTest.php
index 99d4d24174..9200d1f34a 100644
--- a/tests/Console/Scheduling/ScheduleRunCommandTest.php
+++ b/tests/Console/Scheduling/ScheduleRunCommandTest.php
@@ -18,11 +18,13 @@
use Hypervel\Console\Scheduling\Schedule;
use Hypervel\Context\CoroutineContext;
use Hypervel\Contracts\Cache\Repository as Cache;
+use Hypervel\Contracts\Console\Kernel;
use Hypervel\Contracts\Container\Container as ContainerContract;
use Hypervel\Contracts\Debug\ExceptionHandler;
use Hypervel\Contracts\Events\Dispatcher;
use Hypervel\Coroutine\Concurrent;
use Hypervel\Coroutine\Coroutine as HypervelCoroutine;
+use Hypervel\Coroutine\Exceptions\ChildCancellationException;
use Hypervel\Engine\Channel;
use Hypervel\Log\Context\Repository as ContextRepository;
use Hypervel\Support\Carbon;
@@ -38,11 +40,13 @@
use ReflectionProperty;
use RuntimeException;
use Swoole\Coroutine;
+use Swoole\Coroutine\CanceledException;
use Symfony\Component\Console\Input\ArrayInput;
use Symfony\Component\Console\Output\BufferedOutput;
use Throwable;
use function Hypervel\Coroutine\parallel;
+use function Hypervel\Support\defer;
class ScheduleRunCommandTest extends TestCase
{
@@ -91,6 +95,242 @@ public function testForegroundCallbackDispatchesStartingAndFinishedEvents(): voi
$this->assertIsFloat($this->dispatched[1]->runtime);
}
+ public function testScheduledCommandDefersNestedCommandWorkUntilTheTaskFinishes(): void
+ {
+ $calls = [];
+ $kernel = $this->app->make(Kernel::class);
+ $kernel->command('test:deferred-child', function () use (&$calls): void {
+ $calls[] = 'child';
+ defer(function () use (&$calls): void {
+ $calls[] = 'deferred child';
+ });
+ });
+ $kernel->command('test:deferred-parent', function () use (&$calls): void {
+ $calls[] = 'parent';
+ defer(function () use (&$calls): void {
+ $calls[] = 'deferred parent';
+ defer(function () use (&$calls): void {
+ $calls[] = 'deferred during cleanup';
+ });
+ });
+ $this->call('test:deferred-child');
+ $calls[] = 'parent finished';
+ });
+ $event = new Event(m::mock(EventMutex::class), 'test:deferred-parent');
+ $event->after(function () use (&$calls): void {
+ $calls[] = 'after';
+ });
+
+ $this->invokeRunEvents($this->makeCommand(), [$event]);
+
+ $this->assertSame(['parent', 'child', 'parent finished', 'after', 'deferred parent', 'deferred child'], $calls);
+ }
+
+ #[DataProvider('deferredTaskOutcomes')]
+ public function testDeferredCallbacksRespectTheWholeTaskOutcome(string $outcome, array $expected): void
+ {
+ $calls = [];
+ $exception = new RuntimeException('Task failed');
+ $event = new CallbackEvent(m::mock(EventMutex::class), function () use (&$calls, $outcome, $exception): bool {
+ defer(function () use (&$calls): void {
+ $calls[] = 'ordinary';
+ });
+ defer(function () use (&$calls): void {
+ $calls[] = 'always';
+ })->always();
+
+ if ($outcome === 'throw') {
+ throw $exception;
+ }
+
+ return $outcome !== 'false';
+ });
+ if ($outcome === 'after') {
+ $event->after(function () use ($exception): void {
+ throw $exception;
+ });
+ }
+ if ($outcome !== 'success') {
+ $this->handler->shouldReceive('report')->once();
+ }
+ $nextEvent = new CallbackEvent(m::mock(EventMutex::class), function () use (&$calls): void {
+ defer(function () use (&$calls): void {
+ $calls[] = 'next task';
+ });
+ });
+
+ $this->invokeRunEvents($this->makeCommand(), [$event, $nextEvent]);
+
+ $this->assertSame([...$expected, 'next task'], $calls);
+ }
+
+ /**
+ * Provide successful and handled task failures.
+ */
+ public static function deferredTaskOutcomes(): array
+ {
+ return [
+ 'success' => ['success', ['ordinary', 'always']],
+ 'false return' => ['false', ['always']],
+ 'thrown callback' => ['throw', ['always']],
+ 'after callback failure' => ['after', ['always']],
+ ];
+ }
+
+ public function testCanceledCallbackReleasesItsMutexWithoutRunningFailureOrDeferredCallbacks(): void
+ {
+ $calls = [];
+ $cancellation = new CanceledException('Task canceled');
+ $mutex = m::mock(EventMutex::class);
+ $mutex->shouldReceive('exists')->once()->andReturnFalse();
+ $mutex->shouldReceive('create')->once()->andReturnTrue();
+ $mutex->shouldReceive('forget')->once();
+ $event = new CallbackEvent($mutex, function () use (&$calls, $cancellation): void {
+ defer(function () use (&$calls): void {
+ $calls[] = 'always';
+ })->always();
+
+ throw $cancellation;
+ });
+ $event->name('canceled')->withoutOverlapping()->onFailure(function () use (&$calls): void {
+ $calls[] = 'failure';
+ });
+ $this->handler->shouldNotReceive('report');
+
+ try {
+ $this->invokeRunEvents($this->makeCommand(), [$event]);
+ $this->fail('Expected the owned task to report cancellation.');
+ } catch (ChildCancellationException $exception) {
+ $this->assertSame($cancellation, $exception->getPrevious());
+ }
+
+ $this->assertSame([], $calls);
+ $this->assertCount(1, $this->dispatched);
+ $this->assertInstanceOf(ScheduledTaskStarting::class, $this->dispatched[0]);
+ }
+
+ public function testStartingListenerFailureRunsOnlyAlwaysCallbacksAndPreservesTheException(): void
+ {
+ $calls = [];
+ $exception = new RuntimeException('Starting listener failed');
+ $this->dispatcher = $this->app->make(Dispatcher::class);
+ $this->dispatcher->listen(ScheduledTaskStarting::class, function () use (&$calls, $exception): void {
+ defer(function () use (&$calls): void {
+ $calls[] = 'ordinary';
+ });
+ defer(function () use (&$calls): void {
+ $calls[] = 'always';
+ })->always();
+
+ throw $exception;
+ });
+ $event = new CallbackEvent(m::mock(EventMutex::class), static fn (): bool => true);
+ $caught = null;
+
+ try {
+ $this->invokeRunEvents($this->makeCommand(), [$event]);
+ } catch (RuntimeException $throwable) {
+ $caught = $throwable;
+ }
+
+ $this->assertSame($exception, $caught);
+ $this->assertSame(['always'], $calls);
+ }
+
+ public function testCancellationDuringDeferredWorkStopsTheRemainingCallbacks(): void
+ {
+ $calls = [];
+ $cancellation = new CanceledException('Deferred work canceled');
+ $event = new CallbackEvent(m::mock(EventMutex::class), function () use (&$calls, $cancellation): void {
+ defer(function () use ($cancellation): void {
+ throw $cancellation;
+ });
+ defer(function () use (&$calls): void {
+ $calls[] = 'always';
+ })->always();
+ });
+
+ try {
+ $this->invokeRunEvents($this->makeCommand(), [$event]);
+ $this->fail('Expected cancellation from deferred work.');
+ } catch (ChildCancellationException $exception) {
+ $this->assertSame($cancellation, $exception->getPrevious());
+ }
+
+ $this->assertSame([], $calls);
+ }
+
+ public function testBackgroundTaskDrainsItsCallbacksAfterBackgroundFinishedListeners(): void
+ {
+ $calls = [];
+ $this->app->make(Kernel::class)->command('test:background-defer', function () use (&$calls): void {
+ $calls[] = 'command';
+ defer(function () use (&$calls): void {
+ $calls[] = 'deferred command';
+ });
+ });
+ $this->dispatcher = $this->app->make(Dispatcher::class);
+ $this->dispatcher->listen(ScheduledBackgroundTaskFinished::class, function () use (&$calls): void {
+ $calls[] = 'background finished';
+ defer(function () use (&$calls): void {
+ $calls[] = 'deferred listener';
+ });
+ });
+ $event = new Event(m::mock(EventMutex::class), 'test:background-defer');
+ $event->runInBackground();
+ $command = $this->makeCommand();
+ $concurrent = new Concurrent(1);
+ (new ReflectionProperty($command, 'concurrent'))->setValue($command, $concurrent);
+
+ try {
+ $this->invokeRunEvents($command, [$event]);
+ } finally {
+ $this->waitForConcurrent($concurrent);
+ }
+
+ $this->assertSame(['command', 'background finished', 'deferred command', 'deferred listener'], $calls);
+ }
+
+ #[DataProvider('skippedBackgroundTasks')]
+ public function testSkippedBackgroundTaskDoesNotDispatchCompletion(bool $onAnotherServer): void
+ {
+ $mutex = m::mock(EventMutex::class);
+ $event = new Event($mutex, 'test:skipped-background');
+ $event->runInBackground();
+ $command = $this->makeCommand();
+ if ($onAnotherServer) {
+ $event->onOneServer();
+ $schedule = m::mock(Schedule::class);
+ $schedule->shouldReceive('serverShouldRun')->once()->with($event, m::type(CarbonInterface::class))->andReturnFalse();
+ (new ReflectionProperty($command, 'schedule'))->setValue($command, $schedule);
+ } else {
+ $event->withoutOverlapping();
+ $mutex->shouldReceive('exists')->once()->andReturnFalse();
+ $mutex->shouldReceive('create')->once()->andReturnFalse();
+ }
+ $concurrent = new Concurrent(1);
+ (new ReflectionProperty($command, 'concurrent'))->setValue($command, $concurrent);
+
+ try {
+ $this->invokeRunEvents($command, [$event]);
+ } finally {
+ $this->waitForConcurrent($concurrent);
+ }
+
+ $this->assertSame([], array_values(array_filter(
+ $this->dispatched,
+ static fn (object $event): bool => $event instanceof ScheduledBackgroundTaskFinished,
+ )));
+ }
+
+ /**
+ * Provide the two background dispatch skips.
+ */
+ public static function skippedBackgroundTasks(): array
+ {
+ return ['overlapping' => [false], 'another server' => [true]];
+ }
+
public function testFinishedOutputUsesTheSharedRuntimeFormatterWithoutAppendingUnits(): void
{
$event = new ScheduleRunExitCodeEvent(m::mock(EventMutex::class), 0, 'test:duration');
@@ -500,12 +740,19 @@ public function testBackgroundFinishedEventIsNotDispatchedWithoutListeners(): vo
public function testSkippedNonRepeatableTaskIsOnlyEvaluatedOncePerMinute(): void
{
+ $deferred = [];
$eventMutex = m::mock(EventMutex::class);
$callbackEvent = new CallbackEvent($eventMutex, function () {
return 0;
});
- $callbackEvent->when(false);
+ $callbackEvent->when(function () use (&$deferred): bool {
+ defer(function () use (&$deferred): void {
+ $deferred[] = 'filter';
+ });
+
+ return false;
+ });
$command = $this->makeCommand();
$startedAt = CarbonImmutable::parse('2026-05-28 12:34:00');
@@ -516,6 +763,7 @@ public function testSkippedNonRepeatableTaskIsOnlyEvaluatedOncePerMinute(): void
$this->assertCount(1, $this->dispatched);
$this->assertInstanceOf(ScheduledTaskSkipped::class, $this->dispatched[0]);
$this->assertSame($callbackEvent, $this->dispatched[0]->task);
+ $this->assertSame(['filter'], $deferred);
}
public function testSkippedTaskEventIsGuardedByRegisteredListeners(): void
@@ -561,6 +809,28 @@ public function testPausedTaskIsSkippedWithoutRunningFilters(): void
$this->assertSame($callbackEvent, $this->dispatched[0]->task);
}
+ public function testPausedTaskListenerDefersWorkWithinItsOwnCoroutine(): void
+ {
+ $calls = [];
+ $parentCoroutine = Coroutine::getCid();
+ $this->dispatcher = $this->app->make(Dispatcher::class);
+ $this->dispatcher->listen(ScheduledTaskSkipped::class, function () use (&$calls): void {
+ $calls[] = Coroutine::getCid();
+ defer(function () use (&$calls): void {
+ $calls[] = 'deferred';
+ });
+ });
+ $cache = m::mock(Cache::class);
+ $cache->shouldReceive('get')->once()->with('hypervel:schedule:paused', false)->andReturnTrue();
+ $event = new CallbackEvent(m::mock(EventMutex::class), static fn (): bool => true);
+
+ $this->invokeRunEvents($this->makeCommand($cache), [$event]);
+
+ $this->assertCount(2, $calls);
+ $this->assertNotSame($parentCoroutine, $calls[0]);
+ $this->assertSame('deferred', $calls[1]);
+ }
+
public function testTaskMarkedEvenWhenPausedRunsWhileSchedulerIsPaused(): void
{
$runCount = 0;
From 46448d7754bcc5e26d9f12e6064e8a02c06af4d0 Mon Sep 17 00:00:00 2001
From: Raj Siva-Rajah <5361908+binaryfire@users.noreply.github.com>
Date: Sat, 12 Sep 2026 15:38:07 +0000
Subject: [PATCH 2/9] Clarify scheduled task lifecycles and process isolation
Document that scheduled Artisan commands and closures execute inside the scheduler process, with worker-lived static and singleton state. Point commands that require a fresh process to Schedule::exec instead of implying subprocess isolation for command(). Keep the Laravel porting guidance focused on the adaptation required.
Include scheduled tasks in the deferred-functions success policy now that the scheduler owns their cleanup. Documentation was checked against the reviewed implementation and existing command and shell examples.
---
src/docs/helpers.md | 2 +-
src/docs/porting-from-laravel.md | 6 ++++++
src/docs/scheduling.md | 2 ++
3 files changed, 9 insertions(+), 1 deletion(-)
diff --git a/src/docs/helpers.md b/src/docs/helpers.md
index 63bbe0af8d..7c25937430 100644
--- a/src/docs/helpers.md
+++ b/src/docs/helpers.md
@@ -3522,7 +3522,7 @@ Route::post('/orders', function (Request $request) {
});
```
-By default, deferred functions will only be executed if the HTTP response, Artisan command, or queued job from which `Hypervel\Support\defer` is invoked completes successfully. This means that deferred functions will not be executed if a request results in a `4xx` or `5xx` HTTP response. If you would like a deferred function to always execute, you may chain the `always` method onto your deferred function:
+By default, deferred functions will only be executed if the HTTP response, Artisan command, scheduled task, or queued job from which `Hypervel\Support\defer` is invoked completes successfully. This means that deferred functions will not be executed if a request results in a `4xx` or `5xx` HTTP response. If you would like a deferred function to always execute, you may chain the `always` method onto your deferred function:
```php
defer(fn () => Metrics::reportOrder($order))->always();
diff --git a/src/docs/porting-from-laravel.md b/src/docs/porting-from-laravel.md
index 1751d0133c..cef82cd009 100644
--- a/src/docs/porting-from-laravel.md
+++ b/src/docs/porting-from-laravel.md
@@ -21,6 +21,7 @@
- [Coroutine-Aware Dependencies](#coroutine-aware-dependencies)
- [Configuration](#configuration)
- [Other API Differences](#other-api-differences)
+ - [Scheduling](#scheduling)
- [HTTP Client and Concurrency](#http-client-and-concurrency)
- [CSRF Protection](#csrf-protection)
- [Scout](#scout)
@@ -468,6 +469,11 @@ Application code should keep request-specific values in the request, session, co
Many Laravel APIs have direct Hypervel equivalents under the `Hypervel` namespace. The following differences commonly require more than a namespace replacement.
+
+### Scheduling
+
+Scheduled Artisan commands share the scheduler process instead of starting a fresh process for each invocation. Use `exec('php artisan ...')` for commands that rely on process isolation. See [Scheduling Artisan Commands](/docs/{{version}}/scheduling#scheduling-artisan-commands).
+
### HTTP Client and Concurrency
diff --git a/src/docs/scheduling.md b/src/docs/scheduling.md
index 72981e67a4..4f7f13df9b 100644
--- a/src/docs/scheduling.md
+++ b/src/docs/scheduling.md
@@ -91,6 +91,8 @@ Schedule::command('emails:send Taylor --force')->daily();
Schedule::command(SendEmailsCommand::class, ['Taylor', '--force'])->daily();
```
+Scheduled commands and closures run within the scheduler process, so static and singleton state persists between tasks. If a command needs a separate process, schedule it with `Schedule::exec('php artisan emails:send Taylor --force')` instead.
+
#### Scheduling Artisan Closure Commands
From dc7b953bd4e7aa97764f103c2b5f3e67627a73be Mon Sep 17 00:00:00 2001
From: Raj Siva-Rajah <5361908+binaryfire@users.noreply.github.com>
Date: Sat, 12 Sep 2026 16:59:46 +0000
Subject: [PATCH 3/9] Complete deferred work and route termination for
WebSocket callbacks
WebSocket callbacks did not drain deferred functions, and handshake routing never terminated route middleware. Run cleanup at the owning lifecycle boundaries: after onOpen for accepted handshakes, after response emission for uncommitted handshakes, and after message and close lifecycle events.
Keep connection publication atomic before opening, and retain connection context until close cleanup finishes. Use rendered response status and unhandled lifecycle failures to select ordinary deferred callbacks; always callbacks still run after ordinary failures. Cancellation skips remaining deferred work and stays contained at native callback boundaries. Resolve an existing scoped callback collection without allocating one on unused paths, and declare the direct split-package container dependency.
Cover accepted and rejected handshakes, rendered redirects, termination failures, cancellation, event ordering and context release. Formatting, full source/type analysis, affected WebSocket and Reverb suites, the Sentry WebSocket context test and native Reverb integration pass. Regression cases reject the prior implementation. The empty-message benchmark adds about 0.35 microseconds without allocating a callback collection.
---
src/websocket-server/composer.json | 1 +
src/websocket-server/src/Server.php | 141 +++++++++++--
tests/WebSocketServer/ServerHandshakeTest.php | 185 +++++++++++++++++-
tests/WebSocketServer/ServerTest.php | 175 ++++++++++++++++-
4 files changed, 478 insertions(+), 24 deletions(-)
diff --git a/src/websocket-server/composer.json b/src/websocket-server/composer.json
index c42dc6e38e..3bf4f95ff7 100644
--- a/src/websocket-server/composer.json
+++ b/src/websocket-server/composer.json
@@ -32,6 +32,7 @@
"php": "^8.4",
"ext-swoole": "^6.2.2",
"hypervel/collections": "^0.4",
+ "hypervel/container": "^0.4",
"hypervel/context": "^0.4",
"hypervel/contracts": "^0.4",
"hypervel/coordinator": "^0.4",
diff --git a/src/websocket-server/src/Server.php b/src/websocket-server/src/Server.php
index b6c47e955f..5b6217fed9 100644
--- a/src/websocket-server/src/Server.php
+++ b/src/websocket-server/src/Server.php
@@ -4,6 +4,7 @@
namespace Hypervel\WebSocketServer;
+use Hypervel\Container\Container as BaseContainer;
use Hypervel\Context\CoroutineContext;
use Hypervel\Context\RequestContext;
use Hypervel\Contracts\Container\Container;
@@ -27,6 +28,8 @@
use Hypervel\HttpServer\RequestBridge;
use Hypervel\HttpServer\ResponseBridge;
use Hypervel\Routing\Router;
+use Hypervel\Support\Defer\DeferredCallback;
+use Hypervel\Support\Defer\DeferredCallbackCollection;
use Hypervel\Support\SafeCaller;
use Hypervel\WebSocketServer\Collector\FdCollector;
use Hypervel\WebSocketServer\Context as WebSocketContext;
@@ -210,25 +213,38 @@ public function onHandshake(Request $request, SwooleResponse $response): void
}
}
- if ($terminalException !== null) {
- throw $terminalException;
+ if ($terminalException === null && $handshake !== null) {
+ [$fd, $class, $instance, $server] = $handshake;
+
+ if ($server->isEstablished($fd)) {
+ // No yield may occur between the native liveness check and
+ // publication, or onClose could miss the committed connection.
+ $this->deferOnOpen($request, $instance, $server, $fd, $httpRequest, $httpResponse);
+ FdCollector::set($fd, $class);
+ $committed = true;
+
+ return;
+ }
}
- if ($handshake === null) {
- return;
+ if ($httpRequest !== null) {
+ try {
+ $this->terminateRouteMiddleware($httpRequest, $httpResponse);
+ } catch (CanceledException $exception) {
+ throw $exception;
+ } catch (Throwable $throwable) {
+ $terminalException ??= $throwable;
+ }
}
- [$fd, $class, $instance, $server] = $handshake;
+ $this->invokeDeferredCallbacks(
+ $terminalException === null
+ && $httpResponse->getStatusCode() < Response::HTTP_BAD_REQUEST
+ );
- if (! $server->isEstablished($fd)) {
- return;
+ if ($terminalException !== null) {
+ throw $terminalException;
}
-
- // No yield may occur between the native liveness check and
- // publication, or onClose could miss the committed connection.
- $this->deferOnOpen($request, $instance, $server, $fd);
- FdCollector::set($fd, $class);
- $committed = true;
} finally {
if ($fd !== null && ! $committed) {
FdCollector::del($fd);
@@ -305,8 +321,11 @@ public function onMessage(WebSocketServer $server, Frame $frame): void
} catch (CanceledException) {
return;
} catch (Throwable $throwable) {
+ $exception ??= $throwable;
$this->reportCallbackFailure($throwable);
}
+
+ $this->invokeDeferredCallbacks($exception === null);
}
/**
@@ -315,6 +334,7 @@ public function onMessage(WebSocketServer $server, Frame $frame): void
public function onClose(SwooleServer $server, int $fd, int $reactorId): void
{
CoroutineContext::set(WebSocketContext::FD, $fd);
+ $failed = false;
try {
$class = FdCollector::get($fd);
@@ -329,6 +349,7 @@ public function onClose(SwooleServer $server, int $fd, int $reactorId): void
} catch (CanceledException) {
return;
} catch (Throwable $throwable) {
+ $failed = true;
$this->reportCallbackFailure($throwable);
}
@@ -337,6 +358,7 @@ public function onClose(SwooleServer $server, int $fd, int $reactorId): void
} catch (CanceledException) {
return;
} catch (Throwable $throwable) {
+ $failed = true;
$this->reportCallbackFailure($throwable);
}
@@ -345,6 +367,7 @@ public function onClose(SwooleServer $server, int $fd, int $reactorId): void
} catch (CanceledException) {
return;
} catch (Throwable $throwable) {
+ $failed = true;
$this->reportCallbackFailure($throwable);
$instance = null;
}
@@ -355,6 +378,7 @@ public function onClose(SwooleServer $server, int $fd, int $reactorId): void
} catch (CanceledException) {
return;
} catch (Throwable $throwable) {
+ $failed = true;
$this->reportCallbackFailure($throwable);
}
}
@@ -366,8 +390,11 @@ public function onClose(SwooleServer $server, int $fd, int $reactorId): void
} catch (CanceledException) {
return;
} catch (Throwable $throwable) {
+ $failed = true;
$this->reportCallbackFailure($throwable);
}
+
+ $this->invokeDeferredCallbacks(! $failed);
} finally {
FdCollector::del($fd);
WebSocketContext::release($fd);
@@ -457,11 +484,19 @@ protected function getFd(SwooleResponse $response): int
}
/**
- * Defer the onOpen callback after handshake completes.
+ * Defer connection opening and cleanup until the handshake coroutine exits.
*/
- protected function deferOnOpen(Request $request, object $instance, WebSocketServer $server, int $fd): void
- {
- Coroutine::defer(function () use ($request, $instance, $server, $fd) {
+ protected function deferOnOpen(
+ Request $request,
+ object $instance,
+ WebSocketServer $server,
+ int $fd,
+ HttpRequest $httpRequest,
+ Response $httpResponse,
+ ): void {
+ Coroutine::defer(function () use ($request, $instance, $server, $fd, $httpRequest, $httpResponse) {
+ $failed = false;
+
try {
if ($this->event?->hasListeners(ConnectionOpened::class)) {
$this->event->dispatch(new ConnectionOpened($fd, $request, $this->serverName));
@@ -469,6 +504,7 @@ protected function deferOnOpen(Request $request, object $instance, WebSocketServ
} catch (CanceledException) {
return;
} catch (Throwable $throwable) {
+ $failed = true;
$this->reportCallbackFailure($throwable);
}
@@ -478,12 +514,83 @@ protected function deferOnOpen(Request $request, object $instance, WebSocketServ
} catch (CanceledException) {
return;
} catch (Throwable $throwable) {
+ $failed = true;
$this->reportCallbackFailure($throwable);
}
}
+
+ // Termination may yield and let onClose run, so run it after onOpen
+ // has initialized the connection. Drain handshake defers afterward.
+ try {
+ $this->terminateRouteMiddleware($httpRequest, $httpResponse);
+ } catch (CanceledException) {
+ return;
+ } catch (Throwable $throwable) {
+ $failed = true;
+ $this->reportCallbackFailure($throwable);
+ }
+
+ $this->invokeDeferredCallbacks(! $failed);
});
}
+ /**
+ * Terminate the middleware used by the handshake route.
+ */
+ private function terminateRouteMiddleware(HttpRequest $request, Response $response): void
+ {
+ $route = $request->route();
+
+ if ($route === null
+ || ($this->container->bound('middleware.disable')
+ && $this->container->make('middleware.disable') === true)) {
+ return;
+ }
+
+ $exception = null;
+
+ foreach ($this->getRouter()->gatherRouteMiddleware($route) as $middleware) {
+ if (! is_string($middleware)) {
+ continue;
+ }
+
+ try {
+ $instance = $this->container->make(explode(':', $middleware, 2)[0]);
+
+ if (method_exists($instance, 'terminate')) {
+ $instance->terminate($request, $response);
+ }
+ } catch (CanceledException $throwable) {
+ throw $throwable;
+ } catch (Throwable $throwable) {
+ $exception ??= $throwable;
+ }
+ }
+
+ if ($exception !== null) {
+ throw $exception;
+ }
+ }
+
+ /**
+ * Run deferred work without escaping the native callback boundary.
+ */
+ private function invokeDeferredCallbacks(bool $successful): void
+ {
+ try {
+ $container = BaseContainer::getInstance();
+
+ if ($container->resolvedScoped(DeferredCallbackCollection::class)) {
+ $container->make(DeferredCallbackCollection::class)->invokeWhen(
+ fn (DeferredCallback $callback): bool => $successful || $callback->always
+ );
+ }
+ } catch (CanceledException) {
+ } catch (Throwable $throwable) {
+ $this->reportCallbackFailure($throwable);
+ }
+ }
+
/**
* Report a WebSocket callback failure without escaping the native boundary.
*/
diff --git a/tests/WebSocketServer/ServerHandshakeTest.php b/tests/WebSocketServer/ServerHandshakeTest.php
index e1ea5b77ac..d5decca08e 100644
--- a/tests/WebSocketServer/ServerHandshakeTest.php
+++ b/tests/WebSocketServer/ServerHandshakeTest.php
@@ -5,10 +5,14 @@
namespace Hypervel\Tests\WebSocketServer;
use Closure;
+use Hypervel\Container\Container as BaseContainer;
use Hypervel\Contracts\Container\Container;
+use Hypervel\Contracts\Debug\ExceptionHandler;
use Hypervel\Contracts\Log\StdoutLoggerInterface;
+use Hypervel\Contracts\Server\OnOpenInterface;
use Hypervel\Coordinator\Constants;
use Hypervel\Coordinator\CoordinatorManager;
+use Hypervel\Coroutine\Waiter;
use Hypervel\Events\Dispatcher;
use Hypervel\Http\Request as HttpRequest;
use Hypervel\HttpServer\Events\RequestHandled;
@@ -16,15 +20,18 @@
use Hypervel\HttpServer\Events\ResponseSent;
use Hypervel\Routing\Route;
use Hypervel\Routing\Router;
+use Hypervel\Support\Defer\DeferredCallbackCollection;
use Hypervel\Support\SafeCaller;
use Hypervel\Tests\TestCase;
use Hypervel\Tests\WebSocketServer\Fixtures\WebSocketStub;
use Hypervel\WebSocketServer\Collector\FdCollector;
use Hypervel\WebSocketServer\Context as WebSocketContext;
use Hypervel\WebSocketServer\Events\ConnectionOpening;
+use Hypervel\WebSocketServer\Exceptions\Handler\WebSocketExceptionHandler;
use Hypervel\WebSocketServer\Security;
use Hypervel\WebSocketServer\Server;
use Mockery as m;
+use PHPUnit\Framework\Attributes\DataProvider;
use RuntimeException;
use Swoole\Coroutine\CanceledException;
use Swoole\Http\Request as SwooleRequest;
@@ -33,8 +40,163 @@
use Symfony\Component\HttpFoundation\Response;
use Throwable;
+use function Hypervel\Support\defer;
+
class ServerHandshakeTest extends TestCase
{
+ #[DataProvider('acceptedTerminationOutcomes')]
+ public function testAcceptedHandshakeTerminatesAfterOpenAndDrainsOnce(string $outcome, array $expected): void
+ {
+ $calls = [];
+ $exception = $outcome === 'canceled' ? new CanceledException : new RuntimeException('termination failed');
+ $events = new Dispatcher;
+ $events->listen(RequestReceived::class, function () use (&$calls): void {
+ defer(function () use (&$calls): void { $calls[] = 'handshake deferred'; });
+ });
+ $container = $this->container($events);
+ $reporter = m::mock(ExceptionHandler::class);
+ if ($outcome === 'failed') {
+ $reporter->expects('report')->with($exception);
+ } else {
+ $reporter->shouldNotReceive('report');
+ }
+ $container->shouldReceive('make')->with(ExceptionHandler::class)->andReturn($reporter);
+ $container->shouldReceive('make')->with(Security::class)->andReturn(new Security);
+ $handler = m::mock(OnOpenInterface::class);
+ $handler->expects('onOpen')->andReturnUsing(function () use (&$calls): void {
+ $calls[] = 'open';
+ defer(function () use (&$calls): void { $calls[] = 'open deferred'; });
+ });
+ $container->shouldReceive('make')->with(WebSocketStub::class)->andReturn($handler);
+ $middleware = new HandshakeTerminationStub(function () use (&$calls, $outcome, $exception): void {
+ $calls[] = FdCollector::get(42);
+ $calls[] = 'terminated';
+ defer(function () use (&$calls): void { $calls[] = 'termination deferred'; })->always();
+ if ($outcome !== 'success') {
+ throw $exception;
+ }
+ });
+ $container->shouldReceive('make')->with(HandshakeTerminationStub::class)->andReturn($middleware);
+ $native = m::mock(SwooleWebSocketServer::class);
+ $native->expects('isEstablished')->with(42)->andReturnTrue();
+ $server = new HandshakeLifecycleServer($container, $this->router(101, [HandshakeTerminationStub::class . ':value']), $native);
+ $response = $this->response(101, onEnd: function () use (&$calls): bool {
+ $calls[] = 'sent';
+ return true;
+ });
+
+ (new Waiter(1))->wait(fn () => $server->onHandshake($this->request(), $response));
+
+ $this->assertSame(['sent', 'open', WebSocketStub::class, 'terminated', ...$expected], $calls);
+ }
+
+ /**
+ * Supply termination outcomes after the connection opens.
+ */
+ public static function acceptedTerminationOutcomes(): array
+ {
+ return [
+ ['success', ['handshake deferred', 'open deferred', 'termination deferred']],
+ ['failed', ['termination deferred']],
+ ['canceled', []],
+ ];
+ }
+
+ #[DataProvider('uncommittedTerminationOutcomes')]
+ public function testUncommittedHandshakeTerminatesBeforeDrainingAndReleasesContext(int $status, string $outcome, array $expected): void
+ {
+ $calls = [];
+ $exception = $outcome === 'canceled' ? new CanceledException : new RuntimeException('termination failed');
+ $events = new Dispatcher;
+ $events->listen(RequestReceived::class, function () use (&$calls): void {
+ defer(function () use (&$calls): void { $calls[] = 'normal'; });
+ defer(function () use (&$calls): void { $calls[] = WebSocketContext::get('middleware.value'); })->always();
+ });
+ $container = $this->container($events);
+ $container->shouldReceive('make')->with(Security::class)->andReturn(new Security);
+ $container->shouldReceive('make')->with(HandshakeTerminationStub::class)->andReturn(
+ new HandshakeTerminationStub(function () use (&$calls, $outcome, $exception): void {
+ $calls[] = 'first';
+ if ($outcome !== 'success') {
+ throw $exception;
+ }
+ }),
+ );
+ $container->shouldReceive('make')->with('second.middleware')->andReturn(
+ new HandshakeTerminationStub(function () use (&$calls): void { $calls[] = 'second'; }),
+ );
+ $server = new HandshakeLifecycleServer(
+ $container,
+ $this->router($status, [HandshakeTerminationStub::class, 'second.middleware']),
+ m::mock(SwooleWebSocketServer::class),
+ );
+ $caught = null;
+
+ try {
+ $server->onHandshake($this->request(), $this->response($status, $status === 403 ? 'Forbidden' : ''));
+ } catch (Throwable $throwable) {
+ $caught = $throwable;
+ }
+
+ $this->assertSame($outcome === 'success' ? null : $exception, $caught);
+ $this->assertSame($expected, $calls);
+ $this->assertNull(FdCollector::get(42));
+ $this->assertArrayNotHasKey(42, WebSocketContext::getStorage());
+ }
+
+ /**
+ * Supply response statuses and termination outcomes without an upgrade.
+ */
+ public static function uncommittedTerminationOutcomes(): array
+ {
+ return [
+ [302, 'success', ['first', 'second', 'normal', 'preserved']],
+ [302, 'failed', ['first', 'second', 'preserved']],
+ [302, 'canceled', ['first']],
+ [403, 'success', ['first', 'second', 'preserved']],
+ ];
+ }
+
+ public function testHandledHandshakeExceptionUsesTheRenderedStatusForDeferredWork(): void
+ {
+ $calls = [];
+ $exception = new RuntimeException('rendered as a redirect');
+ $events = new Dispatcher;
+ $events->listen(RequestReceived::class, function () use (&$calls, $exception): void {
+ defer(function () use (&$calls): void { $calls[] = 'deferred'; });
+
+ throw $exception;
+ });
+ $container = $this->container($events);
+ $container->expects('make')->with(SafeCaller::class)->andReturn(new SafeCaller($container));
+ $handler = m::mock(WebSocketExceptionHandler::class);
+ $handler->expects('handle')->with($exception, m::type(Response::class))->andReturn(new Response('', 302));
+ $container->expects('make')->with(WebSocketExceptionHandler::class)->andReturn($handler);
+ $server = new HandshakeLifecycleServer($container, m::mock(Router::class), m::mock(SwooleWebSocketServer::class));
+
+ $server->onHandshake($this->request(), $this->response(302));
+
+ $this->assertSame(['deferred'], $calls);
+ }
+
+ public function testDisabledHandshakeMiddlewareIsNotTerminated(): void
+ {
+ $container = $this->container();
+ $container->shouldReceive('bound')->with('middleware.disable')->andReturnTrue();
+ $container->shouldReceive('make')->with('middleware.disable')->andReturnTrue();
+ $container->shouldReceive('make')->with(Security::class)->andReturn(new Security);
+ $container->shouldNotReceive('make')->with(HandshakeTerminationStub::class);
+ $server = new HandshakeLifecycleServer(
+ $container,
+ $this->router(403, [HandshakeTerminationStub::class]),
+ m::mock(SwooleWebSocketServer::class),
+ );
+
+ $server->onHandshake($this->request(), $this->response(403, 'Forbidden'));
+
+ $this->assertNull(FdCollector::get(42));
+ }
+
public function testDispatchesHttpLifecycleAroundNativeHandshakeEmission(): void
{
$order = [];
@@ -289,6 +451,7 @@ protected function container(?Dispatcher $events = null): Container
$container->shouldReceive('make')->once()->with(StdoutLoggerInterface::class)
->andReturn(m::mock(StdoutLoggerInterface::class)->shouldIgnoreMissing());
$container->shouldReceive('bound')->once()->with('events')->andReturn($events !== null);
+ $container->shouldReceive('bound')->with('middleware.disable')->andReturnFalse()->byDefault();
if ($events !== null) {
$container->shouldReceive('make')->once()->with('events')->andReturn($events);
@@ -300,12 +463,13 @@ protected function container(?Dispatcher $events = null): Container
/**
* Create a router returning the requested handshake status.
*/
- protected function router(int $status): Router
+ protected function router(int $status, array $middleware = []): Router
{
$route = m::mock(Route::class);
$route->shouldReceive('getControllerClass')->andReturn(WebSocketStub::class);
$router = m::mock(Router::class);
+ $router->shouldReceive('gatherRouteMiddleware')->with($route)->andReturn($middleware);
$router->shouldReceive('dispatchToCallback')->once()
->with(m::type(HttpRequest::class), m::type(Closure::class))
->andReturnUsing(function (HttpRequest $request) use ($route, $status): Response {
@@ -368,6 +532,25 @@ protected function setUp(): void
parent::setUp();
CoordinatorManager::until(Constants::WORKER_START)->resume();
+ BaseContainer::getInstance()->scoped(DeferredCallbackCollection::class);
+ }
+}
+
+class HandshakeTerminationStub
+{
+ /**
+ * Create middleware with an observable termination callback.
+ */
+ public function __construct(private Closure $callback)
+ {
+ }
+
+ /**
+ * Run the termination callback.
+ */
+ public function terminate(HttpRequest $request, Response $response): void
+ {
+ ($this->callback)();
}
}
diff --git a/tests/WebSocketServer/ServerTest.php b/tests/WebSocketServer/ServerTest.php
index e043c3c77f..8a478446d4 100644
--- a/tests/WebSocketServer/ServerTest.php
+++ b/tests/WebSocketServer/ServerTest.php
@@ -4,13 +4,20 @@
namespace Hypervel\Tests\WebSocketServer;
+use Hypervel\Container\Container as BaseContainer;
use Hypervel\Context\CoroutineContext;
use Hypervel\Contracts\Container\Container;
use Hypervel\Contracts\Debug\ExceptionHandler as ExceptionHandlerContract;
use Hypervel\Contracts\Events\Dispatcher as EventDispatcherContract;
use Hypervel\Contracts\Log\StdoutLoggerInterface;
+use Hypervel\Contracts\Server\OnCloseInterface;
+use Hypervel\Contracts\Server\OnMessageInterface;
+use Hypervel\Contracts\Server\OnOpenInterface;
use Hypervel\Coroutine\Coroutine;
+use Hypervel\Coroutine\Waiter;
+use Hypervel\Http\Request as HttpRequest;
use Hypervel\Support\ClassInvoker;
+use Hypervel\Support\Defer\DeferredCallbackCollection;
use Hypervel\Tests\TestCase;
use Hypervel\Tests\WebSocketServer\Fixtures\WebSocketMessageStub;
use Hypervel\Tests\WebSocketServer\Fixtures\WebSocketStub;
@@ -25,6 +32,7 @@
use Hypervel\WebSocketServer\Exceptions\Handler\WebSocketExceptionHandler;
use Hypervel\WebSocketServer\Server;
use Mockery as m;
+use PHPUnit\Framework\Attributes\DataProvider;
use RuntimeException;
use Swoole\Coroutine\CanceledException;
use Swoole\Http\Request as SwooleRequest;
@@ -34,8 +42,17 @@
use Symfony\Component\HttpFoundation\Response;
use Throwable;
+use function Hypervel\Support\defer;
+
class ServerTest extends TestCase
{
+ protected function setUp(): void
+ {
+ parent::setUp();
+
+ BaseContainer::getInstance()->scoped(DeferredCallbackCollection::class);
+ }
+
protected function tearDown(): void
{
WebSocketMessageStub::flushState();
@@ -62,7 +79,7 @@ public function testDeferOnOpenCallsOnOpen(): void
// Run deferOnOpen inside a child coroutine so that defer() fires
// when that coroutine exits, before we make our assertions.
$coroutineId = Coroutine::create(function () use ($invoker, $swooleServer) {
- $invoker->deferOnOpen(new SwooleRequest, new WebSocketStub, $swooleServer, 1);
+ $invoker->deferOnOpen(new SwooleRequest, new WebSocketStub, $swooleServer, 1, new HttpRequest, new Response('', 101));
});
$this->waitForCoroutine($coroutineId);
@@ -89,7 +106,7 @@ function (string $message) use (&$logged): bool {
$swooleServer = m::mock(WebSocketSwooleServer::class);
$coroutineId = Coroutine::create(function () use ($invoker, $swooleServer) {
- $invoker->deferOnOpen(new SwooleRequest, new WebSocketThrowingStub, $swooleServer, 1);
+ $invoker->deferOnOpen(new SwooleRequest, new WebSocketThrowingStub, $swooleServer, 1, new HttpRequest, new Response('', 101));
});
$this->waitForCoroutine($coroutineId);
@@ -115,7 +132,7 @@ function (ConnectionOpened $event) use (&$dispatched) {
$swooleServer = m::mock(WebSocketSwooleServer::class);
$coroutineId = Coroutine::create(function () use ($invoker, $swooleServer) {
- $invoker->deferOnOpen(new SwooleRequest, new WebSocketStub, $swooleServer, 1);
+ $invoker->deferOnOpen(new SwooleRequest, new WebSocketStub, $swooleServer, 1, new HttpRequest, new Response('', 101));
});
$this->waitForCoroutine($coroutineId);
@@ -142,7 +159,7 @@ function () use (&$hasListenersChecked): bool {
$swooleServer = m::mock(WebSocketSwooleServer::class);
$coroutineId = Coroutine::create(function () use ($invoker, $swooleServer) {
- $invoker->deferOnOpen(new SwooleRequest, new WebSocketStub, $swooleServer, 1);
+ $invoker->deferOnOpen(new SwooleRequest, new WebSocketStub, $swooleServer, 1, new HttpRequest, new Response('', 101));
});
$this->waitForCoroutine($coroutineId);
@@ -171,7 +188,7 @@ function (ConnectionOpened $event) use (&$dispatched) {
$swooleServer = m::mock(WebSocketSwooleServer::class);
$coroutineId = Coroutine::create(function () use ($invoker, $swooleServer) {
- $invoker->deferOnOpen(new SwooleRequest, new WebSocketThrowingStub, $swooleServer, 1);
+ $invoker->deferOnOpen(new SwooleRequest, new WebSocketThrowingStub, $swooleServer, 1, new HttpRequest, new Response('', 101));
});
$this->waitForCoroutine($coroutineId);
@@ -194,13 +211,159 @@ public function testOnOpenRunsWhenConnectionOpenedEventThrows(): void
$swooleServer = m::mock(WebSocketSwooleServer::class);
$coroutineId = Coroutine::create(function () use ($invoker, $swooleServer) {
- $invoker->deferOnOpen(new SwooleRequest, new WebSocketStub, $swooleServer, 1);
+ $invoker->deferOnOpen(new SwooleRequest, new WebSocketStub, $swooleServer, 1, new HttpRequest, new Response('', 101));
});
$this->waitForCoroutine($coroutineId);
$this->assertNotSame(0, WebSocketStub::$coroutineId);
}
+ #[DataProvider('messageDeferredOutcomes')]
+ public function testMessageDeferredWorkFollowsTheCallbackOutcome(?string $failureAt, bool $cancel, array $expected): void
+ {
+ $calls = [];
+ $exception = $cancel ? new CanceledException : new RuntimeException('message failed');
+ $dispatcher = m::mock(EventDispatcherContract::class);
+ $dispatcher->shouldReceive('hasListeners')->andReturnTrue();
+ $dispatcher->shouldReceive('dispatch')->andReturnUsing(function (object $event) use (&$calls, $failureAt, $exception): void {
+ $phase = $event instanceof MessageReceived ? 'received' : 'handled';
+ $calls[] = $phase;
+
+ if ($phase === 'received') {
+ defer(function () use (&$calls): void { $calls[] = 'deferred'; });
+ defer(function () use (&$calls): void { $calls[] = 'always'; })->always();
+ }
+
+ if ($failureAt === $phase) {
+ throw $exception;
+ }
+ });
+ $handler = m::mock(OnMessageInterface::class);
+ $handler->shouldReceive('onMessage')->andReturnUsing(function () use (&$calls, $failureAt, $exception): void {
+ $calls[] = 'handler';
+ if ($failureAt === 'handler') {
+ throw $exception;
+ }
+ });
+ $container = $this->createContainer(dispatcher: $dispatcher);
+ $container->shouldReceive('make')->with(OnMessageInterface::class)->andReturn($handler);
+ FdCollector::set(1, OnMessageInterface::class);
+ $frame = new Frame;
+ $frame->fd = 1;
+
+ (new Server($container))->onMessage(m::mock(WebSocketSwooleServer::class), $frame);
+
+ $this->assertSame($expected, $calls);
+ }
+
+ /**
+ * Supply successful, failed and canceled message phases.
+ */
+ public static function messageDeferredOutcomes(): array
+ {
+ return [
+ 'success' => [null, false, ['received', 'handler', 'handled', 'deferred', 'always']],
+ 'received failure' => ['received', false, ['received', 'handler', 'handled', 'always']],
+ 'handler failure' => ['handler', false, ['received', 'handler', 'handled', 'always']],
+ 'handled failure' => ['handled', false, ['received', 'handler', 'handled', 'always']],
+ 'received cancellation' => ['received', true, ['received']],
+ 'handler cancellation' => ['handler', true, ['received', 'handler']],
+ 'handled cancellation' => ['handled', true, ['received', 'handler', 'handled']],
+ ];
+ }
+
+ #[DataProvider('callbackFailures')]
+ public function testOpenDeferredWorkRunsAfterTheHandler(bool $fail): void
+ {
+ $calls = [];
+ $dispatcher = m::mock(EventDispatcherContract::class);
+ $dispatcher->shouldReceive('hasListeners')->with(ConnectionOpened::class)->andReturnTrue();
+ $dispatcher->expects('dispatch')->andReturnUsing(function () use (&$calls): void {
+ $calls[] = 'opened';
+ defer(function () use (&$calls): void { $calls[] = 'event deferred'; })->always();
+ });
+ $handler = m::mock(OnOpenInterface::class);
+ $handler->expects('onOpen')->andReturnUsing(function () use (&$calls, $fail): void {
+ $calls[] = 'handler';
+ defer(function () use (&$calls): void { $calls[] = 'handler deferred'; });
+ if ($fail) {
+ throw new RuntimeException('open failed');
+ }
+ });
+ $server = new Server($this->createContainer(dispatcher: $dispatcher));
+ $native = m::mock(WebSocketSwooleServer::class);
+
+ (new Waiter(1))->wait(fn () => (new ClassInvoker($server))->deferOnOpen(
+ new SwooleRequest,
+ $handler,
+ $native,
+ 1,
+ new HttpRequest,
+ new Response('', 101)
+ ));
+
+ $this->assertSame($fail
+ ? ['opened', 'handler', 'event deferred']
+ : ['opened', 'handler', 'event deferred', 'handler deferred'], $calls);
+ }
+
+ #[DataProvider('callbackFailures')]
+ public function testCloseDeferredWorkRunsBeforeConnectionContextIsReleased(bool $fail): void
+ {
+ $calls = [];
+ $dispatcher = m::mock(EventDispatcherContract::class);
+ $dispatcher->shouldReceive('hasListeners')->with(ConnectionClosed::class)->andReturnTrue();
+ $dispatcher->expects('dispatch')->andReturnUsing(function () use (&$calls): void { $calls[] = 'closed'; });
+ $handler = m::mock(OnCloseInterface::class);
+ $handler->expects('onClose')->andReturnUsing(function () use (&$calls, $fail): void {
+ defer(function () use (&$calls): void { $calls[] = 'deferred'; });
+ defer(function () use (&$calls): void { $calls[] = WebSocketContext::get('value'); })->always();
+ if ($fail) {
+ throw new RuntimeException('close failed');
+ }
+ });
+ $container = $this->createContainer(dispatcher: $dispatcher);
+ $container->shouldReceive('make')->with(OnCloseInterface::class)->andReturn($handler);
+ FdCollector::set(1, OnCloseInterface::class);
+ CoroutineContext::set(WebSocketContext::FD, 1);
+ WebSocketContext::set('value', 'context');
+
+ (new Server($container))->onClose(m::mock(SwooleServer::class), 1, 0);
+
+ $this->assertSame($fail ? ['closed', 'context'] : ['closed', 'deferred', 'context'], $calls);
+ $this->assertNull(WebSocketContext::get('value', fd: 1));
+ $this->assertNull(FdCollector::get(1));
+ }
+
+ /**
+ * Supply ordinary callback outcomes.
+ */
+ public static function callbackFailures(): array
+ {
+ return [[false], [true]];
+ }
+
+ public function testCancellationDuringDeferredWorkStillReleasesConnectionContext(): void
+ {
+ $calls = [];
+ $handler = m::mock(OnCloseInterface::class);
+ $handler->expects('onClose')->andReturnUsing(function () use (&$calls): void {
+ defer(static fn () => throw new CanceledException);
+ defer(function () use (&$calls): void { $calls[] = 'later'; })->always();
+ });
+ $container = $this->createContainer();
+ $container->shouldReceive('make')->with(OnCloseInterface::class)->andReturn($handler);
+ FdCollector::set(1, OnCloseInterface::class);
+ CoroutineContext::set(WebSocketContext::FD, 1);
+ WebSocketContext::set('value', 'context');
+
+ (new Server($container))->onClose(m::mock(SwooleServer::class), 1, 0);
+
+ $this->assertSame([], $calls);
+ $this->assertNull(WebSocketContext::get('value', fd: 1));
+ $this->assertNull(FdCollector::get(1));
+ }
+
public function testMessageLifecycleEventsAreDispatchedInOrder(): void
{
$dispatcher = m::mock(EventDispatcherContract::class);
From d26f8cd54e28d236d11d968e72f03f80590bb618 Mon Sep 17 00:00:00 2001
From: Raj Siva-Rajah <5361908+binaryfire@users.noreply.github.com>
Date: Sat, 12 Sep 2026 16:59:59 +0000
Subject: [PATCH 4/9] Preserve coroutine cancellation through Reverb protocol
delivery
Application listeners can yield while protocol handling or broadcasting runs under a coroutine deadline. Broad error catches treated cancellation as a protocol failure, emitted an incorrect error frame, or continued sending to later recipients and channels.
Pass cancellation through protocol opening and message handling, recipient delivery, synchronous channel delivery and internal presence publication. Opening still releases its acquired connection slot before rethrowing. Keep ordinary error continuation and mandatory cleanup semantics, without adding cancellation handling to native fan-out paths that do not yield.
Add real deadline regressions for message and send listeners, plus focused assertions for connection-slot cleanup, exception identity and stopping later delivery. Edited files and the complete Reverb suite pass, as do formatting and full source/type analysis. The regressions fail against the previous source.
---
.../src/Protocols/Pusher/Channels/Channel.php | 3 +
.../src/Protocols/Pusher/EventDispatcher.php | 9 ++
src/reverb/src/Protocols/Pusher/Server.php | 7 ++
tests/Reverb/EventDispatcherTest.php | 88 +++++++++++++++++++
.../Protocols/Pusher/Channels/ChannelTest.php | 36 ++++++++
tests/Reverb/Protocols/Pusher/ServerTest.php | 54 ++++++++++++
6 files changed, 197 insertions(+)
diff --git a/src/reverb/src/Protocols/Pusher/Channels/Channel.php b/src/reverb/src/Protocols/Pusher/Channels/Channel.php
index 773cfdf593..d6d6e1e83d 100644
--- a/src/reverb/src/Protocols/Pusher/Channels/Channel.php
+++ b/src/reverb/src/Protocols/Pusher/Channels/Channel.php
@@ -16,6 +16,7 @@
use Hypervel\Reverb\Servers\Hypervel\Scaling\SubscriptionResult;
use Hypervel\Reverb\Webhooks\Contracts\WebhookDispatcher;
use Hypervel\Reverb\Webhooks\DeferredWebhookManager;
+use Swoole\Coroutine\CanceledException;
use Throwable;
class Channel
@@ -387,6 +388,8 @@ protected function sendToConnections(string $message, ?Connection $except = null
try {
$connection->send($message);
+ } catch (CanceledException $throwable) {
+ throw $throwable;
} catch (Throwable $throwable) {
$exception ??= $throwable;
}
diff --git a/src/reverb/src/Protocols/Pusher/EventDispatcher.php b/src/reverb/src/Protocols/Pusher/EventDispatcher.php
index 055c6a999d..063b78dea5 100644
--- a/src/reverb/src/Protocols/Pusher/EventDispatcher.php
+++ b/src/reverb/src/Protocols/Pusher/EventDispatcher.php
@@ -16,6 +16,7 @@
use Hypervel\Reverb\Servers\Hypervel\Contracts\SharedState;
use Hypervel\Support\Arr;
use RuntimeException;
+use Swoole\Coroutine\CanceledException;
use Swoole\Server;
use Throwable;
@@ -86,6 +87,8 @@ public static function dispatchSynchronously(
if ($channel instanceof CacheChannel) {
app(SharedState::class)->clearCacheMissLock($app->id(), $channel->name());
}
+ } catch (CanceledException $throwable) {
+ throw $throwable;
} catch (Throwable $throwable) {
$exception ??= $throwable;
}
@@ -130,6 +133,8 @@ public static function dispatchInternallySynchronously(Application $app, array $
$payload['channel'] = $channel->name();
$channel->broadcastInternally($payload, $connection);
+ } catch (CanceledException $throwable) {
+ throw $throwable;
} catch (Throwable $throwable) {
$exception ??= $throwable;
}
@@ -173,6 +178,8 @@ public static function dispatchInternalToChannel(Application $app, Channel $chan
try {
$channel->broadcastInternally($payload, $connection);
+ } catch (CanceledException $throwable) {
+ throw $throwable;
} catch (Throwable $throwable) {
$exception = $throwable;
}
@@ -197,6 +204,8 @@ public static function dispatchInternalToChannel(Application $app, Channel $chan
try {
app(PubSubProvider::class)->publish($data);
+ } catch (CanceledException $throwable) {
+ throw $throwable;
} catch (Throwable $throwable) {
$exception = $throwable;
}
diff --git a/src/reverb/src/Protocols/Pusher/Server.php b/src/reverb/src/Protocols/Pusher/Server.php
index 9c46f52e6f..9d6572282f 100644
--- a/src/reverb/src/Protocols/Pusher/Server.php
+++ b/src/reverb/src/Protocols/Pusher/Server.php
@@ -21,6 +21,7 @@
use Hypervel\Reverb\Servers\Hypervel\Contracts\SharedState;
use Hypervel\Support\Str;
use JsonException;
+use Swoole\Coroutine\CanceledException;
use Throwable;
class Server
@@ -67,6 +68,10 @@ public function open(Connection $connection): void
}
}
+ if ($e instanceof CanceledException) {
+ throw $e;
+ }
+
try {
$this->error($connection, $e);
} catch (Throwable $throwable) {
@@ -136,6 +141,8 @@ public function message(Connection $from, string $message): void
if (app('events')->hasListeners(MessageReceived::class)) {
MessageReceived::dispatch($from, $message);
}
+ } catch (CanceledException $exception) {
+ throw $exception;
} catch (Throwable $e) {
$terminateOnLimit = $e instanceof RateLimitExceeded
&& $from->app()->rateLimiting()['terminate_on_limit'];
diff --git a/tests/Reverb/EventDispatcherTest.php b/tests/Reverb/EventDispatcherTest.php
index 8f7214affe..d87bc26a9c 100644
--- a/tests/Reverb/EventDispatcherTest.php
+++ b/tests/Reverb/EventDispatcherTest.php
@@ -4,6 +4,7 @@
namespace Hypervel\Tests\Reverb;
+use Hypervel\Contracts\Debug\ExceptionHandler;
use Hypervel\Reverb\Contracts\ApplicationProvider;
use Hypervel\Reverb\Protocols\Pusher\Channels\CacheChannel;
use Hypervel\Reverb\Protocols\Pusher\Channels\Channel;
@@ -18,7 +19,9 @@
use Hypervel\Reverb\Webhooks\Jobs\WebhookDeliveryJob;
use Hypervel\Support\Facades\Queue;
use Mockery as m;
+use PHPUnit\Framework\Attributes\DataProvider;
use RuntimeException;
+use Swoole\Coroutine\CanceledException;
use Swoole\Server;
class EventDispatcherTest extends ReverbTestCase
@@ -139,6 +142,91 @@ public function testCanBroadcastAnEventForMultipleChannels(): void
EventDispatcher::dispatch(app(ApplicationProvider::class)->findByKey('reverb-key'), ['channels' => ['test-channel-one', 'test-channel-two']]);
}
+ #[DataProvider('synchronousDispatchMethods')]
+ public function testCanceledDispatchStopsBeforeLaterChannelsAndWorkers(string $method, string $broadcast): void
+ {
+ $app = app(ApplicationProvider::class)->findByKey('reverb-key');
+ $cancellation = new CanceledException;
+ $channel = m::mock(Channel::class);
+ $channel->allows('name')->andReturn('first');
+ $channel->expects($broadcast)->andThrow($cancellation);
+ $channels = m::mock(ScopedChannelManager::class);
+ $channels->expects('find')->with('first')->andReturn($channel);
+ $channels->shouldNotReceive('find')->with('later');
+ $manager = m::mock(ChannelManager::class);
+ $manager->expects('for')->with($app)->andReturn($channels);
+ $this->app->instance(ChannelManager::class, $manager);
+ $server = m::mock(Server::class);
+ $server->setting = ['worker_num' => 2];
+ $server->worker_id = 0;
+ $server->shouldNotReceive('sendMessage');
+ $this->app->instance(Server::class, $server);
+ $caught = null;
+
+ try {
+ EventDispatcher::{$method}($app, ['channels' => ['first', 'later']]);
+ } catch (CanceledException $exception) {
+ $caught = $exception;
+ }
+
+ $this->assertSame($cancellation, $caught);
+ }
+
+ /**
+ * Supply the public and internal channel delivery methods.
+ */
+ public static function synchronousDispatchMethods(): array
+ {
+ return [
+ ['dispatchSynchronously', 'broadcast'],
+ ['dispatchInternallySynchronously', 'broadcastInternally'],
+ ];
+ }
+
+ #[DataProvider('internalPublishingModes')]
+ public function testInternalChannelCancellationIsNotReported(bool $publish): void
+ {
+ $app = app(ApplicationProvider::class)->findByKey('reverb-key');
+ $cancellation = new CanceledException;
+ $channel = m::mock(Channel::class);
+ $channel->allows('name')->andReturn('presence-test');
+ $handler = m::mock(ExceptionHandler::class);
+ $handler->shouldNotReceive('report');
+ $this->app->instance(ExceptionHandler::class, $handler);
+ $server = m::mock(Server::class);
+ $server->setting = ['worker_num' => 2];
+ $server->worker_id = 0;
+ $server->shouldNotReceive('sendMessage');
+ $this->app->instance(Server::class, $server);
+
+ if ($publish) {
+ app(ServerProviderManager::class)->withPublishing();
+ $provider = m::mock(PubSubProvider::class);
+ $provider->expects('publish')->andThrow($cancellation);
+ $this->app->instance(PubSubProvider::class, $provider);
+ } else {
+ $channel->expects('broadcastInternally')->andThrow($cancellation);
+ }
+
+ $caught = null;
+
+ try {
+ EventDispatcher::dispatchInternalToChannel($app, $channel, ['event' => 'pusher_internal:member_added']);
+ } catch (CanceledException $exception) {
+ $caught = $exception;
+ }
+
+ $this->assertSame($cancellation, $caught);
+ }
+
+ /**
+ * Supply local and Redis-backed internal delivery.
+ */
+ public static function internalPublishingModes(): array
+ {
+ return [[false], [true]];
+ }
+
public function testSynchronousDispatchAttemptsEveryChannelAndFanOutBeforeThrowing(): void
{
$app = app(ApplicationProvider::class)->findByKey('reverb-key');
diff --git a/tests/Reverb/Protocols/Pusher/Channels/ChannelTest.php b/tests/Reverb/Protocols/Pusher/Channels/ChannelTest.php
index 6c36dbfe69..6a0a2a75a8 100644
--- a/tests/Reverb/Protocols/Pusher/Channels/ChannelTest.php
+++ b/tests/Reverb/Protocols/Pusher/Channels/ChannelTest.php
@@ -5,6 +5,13 @@
namespace Hypervel\Tests\Reverb\Protocols\Pusher\Channels;
use Hypervel\Contracts\Debug\ExceptionHandler;
+use Hypervel\Coroutine\Coroutine;
+use Hypervel\Coroutine\Exceptions\WaitTimeoutException;
+use Hypervel\Coroutine\Waiter;
+use Hypervel\Reverb\Connection;
+use Hypervel\Reverb\Contracts\ApplicationProvider;
+use Hypervel\Reverb\Contracts\WebSocketConnection;
+use Hypervel\Reverb\Events\MessageSent;
use Hypervel\Reverb\Protocols\Pusher\Channels\Channel;
use Hypervel\Reverb\Protocols\Pusher\Contracts\ChannelConnectionManager;
use Hypervel\Reverb\Protocols\Pusher\Contracts\ChannelManager;
@@ -222,6 +229,35 @@ public function testCanBroadcastToAllConnectionsOfAChannel(): void
collect($connections)->each(fn ($connection) => $connection->assertReceived(['foo' => 'bar']));
}
+ public function testTimedOutSendListenerStopsDeliveryToLaterConnections(): void
+ {
+ $app = $this->app->make(ApplicationProvider::class)->findByKey('reverb-key');
+ $transport = m::mock(WebSocketConnection::class);
+ $transport->shouldReceive('id')->andReturn(42);
+ $transport->expects('send')->with('{"foo":"bar"}');
+ $first = new Connection($transport, $app, null);
+ $second = new FakeConnection;
+ $this->channelConnectionManager->add($first, []);
+ $this->channelConnectionManager->add($second, []);
+ $handler = m::mock(ExceptionHandler::class);
+ $handler->shouldNotReceive('report');
+ $this->app->instance(ExceptionHandler::class, $handler);
+ $this->app->make('events')->listen(MessageSent::class, static function (): void {
+ Coroutine::sleep(1);
+ });
+ $channel = new Channel('test-channel');
+ $caught = null;
+
+ try {
+ (new Waiter(0.01))->wait(fn () => $channel->broadcast(['foo' => 'bar']));
+ } catch (WaitTimeoutException $exception) {
+ $caught = $exception;
+ }
+
+ $this->assertInstanceOf(WaitTimeoutException::class, $caught);
+ $second->assertNothingReceived();
+ }
+
public function testBroadcastAttemptsEveryConnectionAndReportsTheFirstFailure(): void
{
$firstFailure = new RuntimeException('first send failed');
diff --git a/tests/Reverb/Protocols/Pusher/ServerTest.php b/tests/Reverb/Protocols/Pusher/ServerTest.php
index 881ecdb130..cdc8e53597 100644
--- a/tests/Reverb/Protocols/Pusher/ServerTest.php
+++ b/tests/Reverb/Protocols/Pusher/ServerTest.php
@@ -6,11 +6,15 @@
use Hypervel\Contracts\Debug\ExceptionHandler;
use Hypervel\Contracts\Foundation\Application as ApplicationContract;
+use Hypervel\Coroutine\Coroutine;
+use Hypervel\Coroutine\Exceptions\WaitTimeoutException;
+use Hypervel\Coroutine\Waiter;
use Hypervel\Reverb\Connection;
use Hypervel\Reverb\Contracts\ApplicationProvider;
use Hypervel\Reverb\Contracts\WebSocketConnection;
use Hypervel\Reverb\Events\ConnectionClosed;
use Hypervel\Reverb\Events\ConnectionEstablished;
+use Hypervel\Reverb\Events\MessageReceived;
use Hypervel\Reverb\Protocols\Pusher\Contracts\ChannelManager;
use Hypervel\Reverb\Protocols\Pusher\EventHandler;
use Hypervel\Reverb\Protocols\Pusher\Exceptions\RateLimitExceeded;
@@ -24,6 +28,7 @@
use Mockery as m;
use PHPUnit\Framework\Attributes\DataProvider;
use RuntimeException;
+use Swoole\Coroutine\CanceledException;
class ServerTest extends ReverbTestCase
{
@@ -52,6 +57,55 @@ public function testCanHandleAConnection(): void
]);
}
+ public function testCanceledOpenReleasesItsSlotWithoutSendingAProtocolError(): void
+ {
+ config()->set('reverb.apps.apps.0.max_connections', 1);
+ $state = m::mock(SharedState::class);
+ $state->expects('acquireConnectionSlot')->with('123456', 1)->andReturnTrue();
+ $state->expects('releaseConnectionSlot')->with('123456');
+ $this->app->instance(SharedState::class, $state);
+ $handler = m::mock(ExceptionHandler::class);
+ $handler->shouldNotReceive('report');
+ $this->app->instance(ExceptionHandler::class, $handler);
+ $cancellation = new CanceledException;
+ Event::listen(ConnectionEstablished::class, static fn () => throw $cancellation);
+ $connection = new FakeConnection;
+ $caught = null;
+
+ try {
+ $this->server->open($connection);
+ } catch (CanceledException $exception) {
+ $caught = $exception;
+ }
+
+ $this->assertSame($cancellation, $caught);
+ $this->assertFalse($connection->hasAcquiredConnectionSlot());
+ $this->assertFalse($connection->isEstablished());
+ $connection->assertReceivedCount(1);
+ $this->assertSame('pusher:connection_established', json_decode($connection->messages[0], true)['event']);
+ }
+
+ public function testTimedOutMessageDoesNotSendOrReportAProtocolError(): void
+ {
+ $handler = m::mock(ExceptionHandler::class);
+ $handler->shouldNotReceive('report');
+ $this->app->instance(ExceptionHandler::class, $handler);
+ Event::listen(MessageReceived::class, static function (): void {
+ Coroutine::sleep(1);
+ });
+ $connection = new FakeConnection;
+ $caught = null;
+
+ try {
+ (new Waiter(0.01))->wait(fn () => $this->server->message($connection, '{"event":"pusher:ping"}'));
+ } catch (WaitTimeoutException $exception) {
+ $caught = $exception;
+ }
+
+ $this->assertInstanceOf(WaitTimeoutException::class, $caught);
+ $this->assertSame(['{"event":"pusher:pong"}'], $connection->messages);
+ }
+
public function testCanHandleADisconnection(): void
{
$scopedManager = m::spy(ScopedChannelManager::class);
From 2b6916faa36a769964b208e03779d39a655cf11f Mon Sep 17 00:00:00 2001
From: Raj Siva-Rajah <5361908+binaryfire@users.noreply.github.com>
Date: Sat, 12 Sep 2026 17:00:17 +0000
Subject: [PATCH 5/9] Finish channel cleanup when a Reverb unsubscription fails
A listener, webhook or presence delivery failure after removing one membership stopped the remaining channel cleanup. Closing had already removed the lifecycle from the connection registry, so later memberships could retain a closed connection indefinitely.
Attempt every channel unsubscription once and rethrow the first failure after cleanup. Include cancellation in this mandatory cleanup aggregation, matching the surrounding connection-close lifecycle. Keep the existing shared-state failure boundary; this does not retry writes or discard an unconfirmed membership change.
Use two real channels to verify ordinary failures and cancellation still remove later memberships and preserve the first exception. Both regression cases fail with the previous loop. The focused manager tests, complete Reverb suite, formatting and full source/type analysis pass. The upstream Laravel Reverb manager contains the same non-aggregating loop and is a candidate for an equivalent upstream correction.
---
.../Pusher/Managers/ArrayChannelManager.php | 14 ++++++-
.../Pusher/Managers/ChannelManagerTest.php | 38 +++++++++++++++++++
2 files changed, 51 insertions(+), 1 deletion(-)
diff --git a/src/reverb/src/Protocols/Pusher/Managers/ArrayChannelManager.php b/src/reverb/src/Protocols/Pusher/Managers/ArrayChannelManager.php
index 6f4f990c92..33df97a21c 100644
--- a/src/reverb/src/Protocols/Pusher/Managers/ArrayChannelManager.php
+++ b/src/reverb/src/Protocols/Pusher/Managers/ArrayChannelManager.php
@@ -11,6 +11,7 @@
use Hypervel\Reverb\Protocols\Pusher\Channels\ChannelConnection;
use Hypervel\Reverb\Protocols\Pusher\Contracts\ChannelManager as ChannelManagerInterface;
use Hypervel\Support\Arr;
+use Throwable;
class ArrayChannelManager implements ChannelManagerInterface
{
@@ -116,8 +117,19 @@ public function findConnection(string $appId, string $socketId): ?ChannelConnect
*/
public function unsubscribeFromAllChannels(string $appId, Connection $connection): void
{
+ $exception = null;
+
+ // A failed unsubscription must not leave the connection in later channels.
foreach ($this->channels($appId) as $channel) {
- $channel->unsubscribe($connection);
+ try {
+ $channel->unsubscribe($connection);
+ } catch (Throwable $throwable) {
+ $exception ??= $throwable;
+ }
+ }
+
+ if ($exception !== null) {
+ throw $exception;
}
}
diff --git a/tests/Reverb/Protocols/Pusher/Managers/ChannelManagerTest.php b/tests/Reverb/Protocols/Pusher/Managers/ChannelManagerTest.php
index 402ddeb3d8..f42d4309b7 100644
--- a/tests/Reverb/Protocols/Pusher/Managers/ChannelManagerTest.php
+++ b/tests/Reverb/Protocols/Pusher/Managers/ChannelManagerTest.php
@@ -4,11 +4,16 @@
namespace Hypervel\Tests\Reverb\Protocols\Pusher\Managers;
+use Hypervel\Reverb\Events\ChannelRemoved;
use Hypervel\Reverb\Protocols\Pusher\Channels\Channel;
use Hypervel\Reverb\Protocols\Pusher\Contracts\ChannelManager;
use Hypervel\Reverb\Protocols\Pusher\Contracts\ScopedChannelManager;
use Hypervel\Tests\Reverb\Fixtures\FakeConnection;
use Hypervel\Tests\Reverb\ReverbTestCase;
+use PHPUnit\Framework\Attributes\DataProvider;
+use RuntimeException;
+use Swoole\Coroutine\CanceledException;
+use Throwable;
class ChannelManagerTest extends ReverbTestCase
{
@@ -90,6 +95,39 @@ public function testCanUnsubscribeAConnectionFromAllChannels(): void
collect($this->channelManager->all())->each(fn ($channel) => $this->assertCount(0, $channel->connections()));
}
+ #[DataProvider('unsubscribeFailures')]
+ public function testUnsubscribeFailureDoesNotLeaveMembershipInLaterChannels(bool $cancel): void
+ {
+ $first = $this->channel;
+ $second = $this->channelManager->findOrCreate('second');
+ $first->subscribe($this->connection);
+ $second->subscribe($this->connection);
+ $failure = $cancel ? new CanceledException : new RuntimeException('listener failed');
+ $laterFailure = new RuntimeException('later listener failed');
+ $this->app->make('events')->listen(ChannelRemoved::class, static function (ChannelRemoved $event) use ($first, $failure, $laterFailure): never {
+ throw $event->channel === $first ? $failure : $laterFailure;
+ });
+ $caught = null;
+
+ try {
+ $this->channelManager->unsubscribeFromAll($this->connection);
+ } catch (Throwable $exception) {
+ $caught = $exception;
+ }
+
+ $this->assertSame($failure, $caught);
+ $this->assertFalse($first->subscribed($this->connection));
+ $this->assertFalse($second->subscribed($this->connection));
+ }
+
+ /**
+ * Supply ordinary failure and cancellation during channel cleanup.
+ */
+ public static function unsubscribeFailures(): array
+ {
+ return [[false], [true]];
+ }
+
public function testCanGetTheDataForAConnectionSubscribedToAChannel(): void
{
collect(static::factory(5))->each(fn ($connection) => $this->channel->subscribe(
From c9bf5980deb3d3ca2cebfd6d2265dd5d9d7c8aa2 Mon Sep 17 00:00:00 2001
From: Raj Siva-Rajah <5361908+binaryfire@users.noreply.github.com>
Date: Sat, 12 Sep 2026 17:00:29 +0000
Subject: [PATCH 6/9] Document WebSocket middleware termination and deferred
functions
Describe when handshake route middleware terminates and when work deferred during opening, message handling or closing runs. Include WebSocket callbacks in the deferred-function success rules and explain that coroutine cancellation skips deferred work.
Keep the public guidance in the existing helper and WebSocket pages. The documented ordering and failure behavior are covered by the WebSocket lifecycle tests.
---
src/docs/helpers.md | 2 +-
src/docs/websockets.md | 2 ++
2 files changed, 3 insertions(+), 1 deletion(-)
diff --git a/src/docs/helpers.md b/src/docs/helpers.md
index 7c25937430..596f43655a 100644
--- a/src/docs/helpers.md
+++ b/src/docs/helpers.md
@@ -3522,7 +3522,7 @@ Route::post('/orders', function (Request $request) {
});
```
-By default, deferred functions will only be executed if the HTTP response, Artisan command, scheduled task, or queued job from which `Hypervel\Support\defer` is invoked completes successfully. This means that deferred functions will not be executed if a request results in a `4xx` or `5xx` HTTP response. If you would like a deferred function to always execute, you may chain the `always` method onto your deferred function:
+By default, deferred functions will only be executed if the HTTP response, Artisan command, scheduled task, WebSocket callback, or queued job from which `Hypervel\Support\defer` is invoked completes successfully. This means that deferred functions will not be executed if a request results in a `4xx` or `5xx` HTTP response. If you would like a deferred function to always execute, you may chain the `always` method onto your deferred function:
```php
defer(fn () => Metrics::reportOrder($order))->always();
diff --git a/src/docs/websockets.md b/src/docs/websockets.md
index a32cd10ab8..33a9bf8dbf 100644
--- a/src/docs/websockets.md
+++ b/src/docs/websockets.md
@@ -122,6 +122,8 @@ class ChatSocket implements OnOpenInterface, OnMessageInterface, OnCloseInterfac
The handler's `__invoke` method handles ordinary HTTP requests because the route is also visible to your application's HTTP server. WebSocket handshakes run the route's middleware but use the handler's lifecycle methods instead of invoking the controller.
+Terminable route middleware runs after `onOpen`, or after a rejected handshake response has been sent. [Deferred functions](/docs/{{version}}/helpers#deferred-functions) registered during the handshake or `onOpen` run after this cleanup. Functions deferred by `onMessage` or `onClose` run after the callback and its lifecycle events finish. They follow the usual success and `always()` rules; coroutine cancellation skips deferred work.
+
Handlers are resolved through the service container and normally live for the lifetime of a worker. Do not store connection-specific data on handler properties.
From 5624f71d5e6680a933cc5b3d068a8fc6ebf8444c Mon Sep 17 00:00:00 2001
From: Raj Siva-Rajah <5361908+binaryfire@users.noreply.github.com>
Date: Sat, 12 Sep 2026 17:27:50 +0000
Subject: [PATCH 7/9] Declare the WebSocket opening callback return type
The deferred connection-opening callback only returns without a value or completes normally. Declare its native void return type to match the framework typing convention.
This leaves connection initialization, middleware termination, deferred callback eligibility, and cancellation handling unchanged.
Validated with the WebSocketServer suite, configured PHP CS Fixer, and full source and type-fixture analysis.
---
src/websocket-server/src/Server.php | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
diff --git a/src/websocket-server/src/Server.php b/src/websocket-server/src/Server.php
index 5b6217fed9..8fd6530e5c 100644
--- a/src/websocket-server/src/Server.php
+++ b/src/websocket-server/src/Server.php
@@ -494,7 +494,7 @@ protected function deferOnOpen(
HttpRequest $httpRequest,
Response $httpResponse,
): void {
- Coroutine::defer(function () use ($request, $instance, $server, $fd, $httpRequest, $httpResponse) {
+ Coroutine::defer(function () use ($request, $instance, $server, $fd, $httpRequest, $httpResponse): void {
$failed = false;
try {
From 415a28cf9e717748563b57e8e7d9d284f161371e Mon Sep 17 00:00:00 2001
From: Raj Siva-Rajah <5361908+binaryfire@users.noreply.github.com>
Date: Sat, 12 Sep 2026 17:27:58 +0000
Subject: [PATCH 8/9] Declare the Telescope deferred storage callback return
type
Telescope schedules a coroutine-exit callback that calls its void storage method and does not return a value. Declare void on that closure, matching the other typed deferred callbacks.
Recording state, storage timing, and exception propagation are unchanged. No new tests are needed for this annotation correction.
Validated with the Telescope suite, configured PHP CS Fixer, and full source and type-fixture analysis.
---
src/telescope/src/Telescope.php | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
diff --git a/src/telescope/src/Telescope.php b/src/telescope/src/Telescope.php
index 301bca287e..ab611cc1fe 100644
--- a/src/telescope/src/Telescope.php
+++ b/src/telescope/src/Telescope.php
@@ -273,7 +273,7 @@ protected static function record(string $type, IncomingEntry $entry): void
if (Coroutine::inCoroutine()
&& ! $state->storeScheduled
) {
- Coroutine::defer(function () {
+ Coroutine::defer(function (): void {
static::store(static::$store);
});
$state->storeScheduled = true;
From 19283275196ee224eeff0cc2dfa63e56bd7a71b8 Mon Sep 17 00:00:00 2001
From: Raj Siva-Rajah <5361908+binaryfire@users.noreply.github.com>
Date: Sat, 12 Sep 2026 17:35:05 +0000
Subject: [PATCH 9/9] Explain cancellation handling at the Reverb worker pipe
Document why native worker delivery retains ordinary error aggregation while local channel delivery rethrows cancellation. The native send does not yield for the framework protocol payloads, so it cannot receive coroutine cancellation at that boundary.
Keep the explanation beside the worker loop so future changes do not mistake the absent cancellation catch for an omission. No executable code or tests change.
Verified the native transport path, the comment-only diff, and configured PHP CS Fixer.
---
src/reverb/src/Protocols/Pusher/EventDispatcher.php | 2 ++
1 file changed, 2 insertions(+)
diff --git a/src/reverb/src/Protocols/Pusher/EventDispatcher.php b/src/reverb/src/Protocols/Pusher/EventDispatcher.php
index 063b78dea5..176efd87dd 100644
--- a/src/reverb/src/Protocols/Pusher/EventDispatcher.php
+++ b/src/reverb/src/Protocols/Pusher/EventDispatcher.php
@@ -244,6 +244,8 @@ protected static function fanOutToOtherWorkers(
);
$exception = null;
+ // Unlike channel delivery, sendMessage() never yields for these payloads,
+ // so it cannot be canceled.
for ($workerId = 0; $workerId < $workerNum; ++$workerId) {
if ($workerId === $currentWorkerId) {
continue;