Complete scheduler and WebSocket lifecycle cleanup

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

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,

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, 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();

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.
 
+<a name="scheduling"></a>
+### 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).
+
 <a name="http-client-and-concurrency"></a>
 ### HTTP Client and Concurrency
 

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.
+
 <a name="scheduling-artisan-closure-commands"></a>
 #### Scheduling Artisan Closure Commands
 

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.
 
 <a name="connection-context"></a>

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;
             }

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;
             }
@@ -235,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;

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;
         }
     }
 

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'];

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;

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",