Make Guzzle coroutine-safe and reduce AOP overhead

.github/workflows/tests.yml

@@ -71,6 +71,43 @@ jobs:
           OTEL_SDK_DISABLED: true
         run: vendor/bin/testbench package:test --parallel
 
+  guzzle_7_compatibility:
+    runs-on: ubuntu-latest
+    timeout-minutes: 15
+    if: "!contains(github.event.head_commit.message, '[skip ci]') && !contains(github.event.head_commit.message, '[ci skip]')"
+    name: Guzzle 7 and promises 2
+
+    container:
+      image: ghcr.io/hypervel/components-ci:php8.4-swoole6.2.2
+
+    steps:
+      - name: Checkout code
+        uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
+        with:
+          persist-credentials: false
+
+      - name: Log trigger context
+        uses: ./.github/actions/log-trigger-context
+
+      - name: Cache Composer dependencies
+        uses: actions/cache@55cc8345863c7cc4c66a329aec7e433d2d1c52a9 # v6.1.0
+        with:
+          path: /root/.composer/cache
+          key: composer-guzzle7-${{ hashFiles('composer.json') }}
+          restore-keys: composer-guzzle7-
+
+      - name: Install the supported Guzzle 7 dependency family
+        run: |
+          COMPOSER_MEMORY_LIMIT=-1 composer update --prefer-dist -n -o \
+            --with 'guzzlehttp/guzzle:^7.15.2' \
+            --with 'guzzlehttp/promises:^2.5.2' \
+            --with 'guzzlehttp/psr7:^2.13'
+
+      - name: Run ownership and affected consumer tests
+        run: >-
+          vendor/bin/paratest -c phpunit.xml.dist
+          --filter '(Guzzle|PusherClientTest|AwsCredentialConsumerTest|SerializedCredentialProvider|HttpConnectionTest|CurlStreamingHandlerTest|HttpClientStreamingTest|AfterEachTestExtensionTest|GenerateProxiesTest|ServiceProviderDiTest)'
+
   wayfinder_tests:
     runs-on: ubuntu-latest
     timeout-minutes: 20

.php-cs-fixer.php

@@ -82,9 +82,8 @@
         'phpdoc_to_comment' => [
             'ignored_tags' => ['var'],
         ],
-        'return_assignment' => [
-            'skip_named_var_tags' => true,
-        ],
+        // This rewrite removes assignments captured by reference in nested closures.
+        'return_assignment' => false,
         'php_unit_method_casing' => [
             'case' => 'camel_case',
         ],

docs/todo.md

@@ -30,6 +30,7 @@
 
 ## Testing
 
+- Integrate the resolution of [PHPUnit #7039](https://github.com/sebastianbergmann/phpunit/issues/7039) into `RunTestsInCoroutine` once available; [PR #7040](https://github.com/sebastianbergmann/phpunit/pull/7040) proposes suspend/resume output buffering. Follow the accepted upstream design, including calling `parent::invokeTestMethod()` if capture is handled there. Add regression coverage for output expectations, flushed output and buffer cleanup; update the testing guidance and reassess output-related coroutine opt-outs.
 - Lift the PHPUnit `13.3.*` pin in the root, Testbench and dogfood package manifests, and in the `hypervel/hypervel` application skeleton, once ParaTest supports PHPUnit 13.4. PHPUnit 13.4.0 made the internal `PhpHandler` constructor require an event emitter, and ParaTest 7.25.0 and Hypervel's `RunsInParallel` still construct it without one, so parallel runs fail before any test starts. When lifting the pin, pass the emitter to `PhpHandler` in `RunsInParallel`, as PHPUnit 13.4's own `Application` does.
 
 ## Filesystem

src/broadcasting/README.md

@@ -11,6 +11,8 @@ The outgoing channel formatter and incoming channel authorizer are also worker-w
 
 Custom drivers may opt into Hypervel's connection pooling through the broadcast manager.
 
+The default Pusher/Reverb client rejects Guzzle's `max_host_connections` and `max_total_connections` options because enforcing them across coroutines requires an unsafe shared multi-handler. Bound broadcast concurrency or use rate limiting instead.
+
 Mercure uses a standalone HTTP hub. FrankenPHP's in-process `mercure_publish()` integration is not available under Swoole.
 
 The broadcast service provider does not implement Laravel's `DeferrableProvider` marker because Hypervel has no deferred service provider mechanism.

src/broadcasting/composer.json

@@ -26,10 +26,12 @@
     "require": {
         "php": "^8.4",
         "guzzlehttp/guzzle": "^7.15.2 || ^8.2",
+        "guzzlehttp/promises": "^2.5.2 || ^3.0.2",
         "hypervel/bus": "^0.4",
         "hypervel/collections": "^0.4",
         "hypervel/connection-pool": "^0.4",
         "hypervel/container": "^0.4",
+        "hypervel/context": "^0.4",
         "hypervel/contracts": "^0.4",
         "hypervel/foundation": "^0.4",
         "hypervel/http": "^0.4",
@@ -38,6 +40,7 @@
         "hypervel/reflection": "^0.4",
         "hypervel/routing": "^0.4",
         "hypervel/support": "^0.4",
+        "psr/http-message": "^2.0",
         "psr/log": "^3.0",
         "symfony/http-kernel": "^8.1"
     },

src/broadcasting/src/BroadcastManager.php

@@ -7,6 +7,10 @@
 use Ably\AblyRest;
 use Closure;
 use GuzzleHttp\Client as GuzzleClient;
+use GuzzleHttp\HandlerStack;
+use GuzzleHttp\Multiplexing;
+use GuzzleHttp\Promise\PromiseInterface;
+use GuzzleHttp\Utils;
 use Hypervel\Broadcasting\Broadcasters\AblyBroadcaster;
 use Hypervel\Broadcasting\Broadcasters\Broadcaster as BaseBroadcaster;
 use Hypervel\Broadcasting\Broadcasters\LogBroadcaster;
@@ -17,6 +21,7 @@
 use Hypervel\Broadcasting\Mercure\CreatesMercureDrivers;
 use Hypervel\Bus\DispatchLockContext;
 use Hypervel\Bus\UniqueLock;
+use Hypervel\Context\CoroutineContext;
 use Hypervel\Contracts\Broadcasting\Broadcaster;
 use Hypervel\Contracts\Broadcasting\Factory as BroadcastingFactoryContract;
 use Hypervel\Contracts\Broadcasting\ShouldBeUnique;
@@ -42,6 +47,7 @@
 use Hypervel\Support\Queue\Concerns\ResolvesQueueRoutes;
 use Hypervel\Support\RebindsCallbacksToSelf;
 use InvalidArgumentException;
+use Psr\Http\Message\RequestInterface;
 use Psr\Log\LoggerInterface;
 use Pusher\Pusher;
 use ReflectionException;
@@ -62,6 +68,8 @@ class BroadcastManager implements BroadcastingFactoryContract
     use RebindsCallbacksToSelf;
     use ResolvesQueueRoutes;
 
+    protected const string PUSHER_HANDLERS_CONTEXT_KEY = '__broadcasting.pusher_handlers';
+
     /**
      * The array of resolved broadcast drivers.
      */
@@ -395,17 +403,51 @@ protected function createPusherDriver(array $config): Broadcaster
      */
     public function pusher(array $config): Pusher
     {
-        $guzzleClient = new GuzzleClient(
-            array_merge(
-                [
-                    'connect_timeout' => 10,
-                    'crypto_method' => STREAM_CRYPTO_METHOD_TLSv1_2_CLIENT,
-                    'timeout' => 30,
-                ],
-                $config['client_options'] ?? [],
-            ),
+        $clientOptions = array_merge(
+            [
+                'connect_timeout' => 10,
+                'crypto_method' => STREAM_CRYPTO_METHOD_TLSv1_2_CLIENT,
+                'timeout' => 30,
+            ],
+            $config['client_options'] ?? [],
         );
 
+        if (! isset($clientOptions['handler'])) {
+            foreach (['max_host_connections', 'max_total_connections'] as $option) {
+                if (isset($clientOptions[$option])) {
+                    throw new InvalidArgumentException(
+                        "The [{$option}] broadcasting client option requires a shared multi-handler, which is unsafe across coroutines. Bound coroutine concurrency or use rate limiting instead."
+                    );
+                }
+            }
+
+            $handlerOptions = Arr::only($clientOptions, ['transport_sharing']);
+            unset($clientOptions['transport_sharing']);
+
+            if (($clientOptions['multiplex'] ?? null) === Multiplexing::NONE) {
+                $handlerOptions['multiplex'] = Multiplexing::NONE;
+            }
+
+            $sharedHandler = Utils::chooseHandler($handlerOptions)(...);
+
+            // Async transfers may share a multi-handler only within their owning coroutine.
+            $clientOptions['handler'] = HandlerStack::create(
+                static function (RequestInterface $request, array $options) use ($sharedHandler, $handlerOptions): PromiseInterface {
+                    if (! empty($options['synchronous'])) {
+                        return $sharedHandler($request, $options);
+                    }
+
+                    $context = CoroutineContext::get(self::PUSHER_HANDLERS_CONTEXT_KEY)
+                        ?? CoroutineContext::set(self::PUSHER_HANDLERS_CONTEXT_KEY, new PusherHandlerContext);
+                    $handler = $context->handlers[$sharedHandler] ??= Utils::chooseHandler($handlerOptions);
+
+                    return $handler($request, $options);
+                },
+            );
+        }
+
+        $guzzleClient = new GuzzleClient($clientOptions);
+
         $pusher = new Pusher(
             $config['key'],
             $config['secret'],

src/broadcasting/src/PusherHandlerContext.php

@@ -0,0 +1,26 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Hypervel\Broadcasting;
+
+use Closure;
+use Hypervel\Context\NonCopyableContext;
+use WeakMap;
+
+/**
+ * Keep async transports in their owning coroutine, including when children copy context.
+ */
+class PusherHandlerContext implements NonCopyableContext
+{
+    /** @var WeakMap<Closure, callable> */
+    public readonly WeakMap $handlers;
+
+    /**
+     * Create a coroutine-local collection of client handlers.
+     */
+    public function __construct()
+    {
+        $this->handlers = new WeakMap;
+    }
+}

src/cache/src/Console/ClearCommand.php

@@ -130,7 +130,7 @@ protected function cache(): Repository
      */
     public function flushProxies(): void
     {
-        $this->files->deleteDirectory(storage_path('framework/aop'));
+        $this->files->deleteDirectory($this->hypervel->bootstrapPath('cache/aop'));
     }
 
     /**

src/di/composer.json

@@ -37,7 +37,6 @@
         "hypervel/collections": "^0.4",
         "hypervel/container": "^0.4",
         "hypervel/filesystem": "^0.4",
-        "hypervel/pipeline": "^0.4",
         "hypervel/reflection": "^0.4",
         "hypervel/support": "^0.4",
         "nikic/php-parser": "^5.7"

src/di/src/Aop/Aspect.php

@@ -47,6 +47,10 @@ public static function isMatchClassRule(string $target, string $rule): array
             [$class, $method] = explode('::', $target);
         }
 
+        if ($ruleMethod === null && in_array($method, RewriteCollection::CLASS_EXCLUDED_METHODS, true)) {
+            return [false, null];
+        }
+
         if ($method === null) {
             if (! str_contains($ruleClass, '*')) {
                 /*
@@ -132,7 +136,7 @@ private static function parseClasses(array $collection, string $class, RewriteCo
                 if ($isMatch) {
                     if ($method === null) {
                         $rewriteCollection->setLevel(RewriteCollection::CLASS_LEVEL);
-                        return;
+                        continue;
                     }
                     $rewriteCollection->add($method);
                 }

src/di/src/Aop/AspectCollector.php

@@ -4,6 +4,9 @@
 
 namespace Hypervel\Di\Aop;
 
+use Hypervel\Support\ClassMetadataCache;
+use ReflectionProperty;
+
 /**
  * Static registry of aspect class rules and priorities.
  *
@@ -21,6 +24,32 @@ class AspectCollector
      */
     protected static array $aspectRules = [];
 
+    /**
+     * Register an aspect using its public default targeting rules and priority.
+     *
+     * Boot-only. Register before proxy generation; loaded classes cannot be rewritten.
+     *
+     * @param class-string $aspect
+     */
+    public static function register(string $aspect): void
+    {
+        $reflectionClass = ClassMetadataCache::reflectClass($aspect);
+        $properties = $reflectionClass->getProperties(ReflectionProperty::IS_PUBLIC);
+
+        $classes = [];
+        $priority = null;
+
+        foreach ($properties as $property) {
+            if ($property->getName() === 'classes') {
+                $classes = $property->getDefaultValue();
+            } elseif ($property->getName() === 'priority') {
+                $priority = $property->getDefaultValue();
+            }
+        }
+
+        static::setAround($aspect, $classes, $priority);
+    }
+
     /**
      * Register an aspect with its class targeting rules.
      *

src/di/src/Aop/AspectManager.php

@@ -4,48 +4,43 @@
 
 namespace Hypervel\Di\Aop;
 
+use Closure;
+
 /**
- * Runtime cache of resolved aspects per class::method.
- *
- * Once the aspect pipeline for a given class method is resolved,
- * it is cached here so subsequent calls skip the resolution logic.
+ * Cache immutable aspect chains for the worker's intercepted methods.
  */
 class AspectManager
 {
     /**
-     * @var array<string, array<string, array<int, string>>>
-     */
-    protected static array $container = [];
-
-    /**
-     * Get the resolved aspects for a class method.
+     * @var array<string, array<string, Closure(ProceedingJoinPoint): mixed>>
      */
-    public static function get(string $class, string $method): array
-    {
-        return static::$container[$class][$method] ?? [];
-    }
+    protected static array $chains = [];
 
     /**
-     * Determine if aspects have been resolved for a class method.
+     * Get the compiled aspect chain for a class method.
+     *
+     * @return null|Closure(ProceedingJoinPoint): mixed
      */
-    public static function has(string $class, string $method): bool
+    public static function get(string $class, string $method): ?Closure
     {
-        return isset(static::$container[$class][$method]);
+        return static::$chains[$class][$method] ?? null;
     }
 
     /**
-     * Set the resolved aspects for a class method.
+     * Set the compiled aspect chain for a class method.
+     *
+     * @param Closure(ProceedingJoinPoint): mixed $value
      */
-    public static function set(string $class, string $method, array $value): void
+    public static function set(string $class, string $method, Closure $value): void
     {
-        static::$container[$class][$method] = $value;
+        static::$chains[$class][$method] = $value;
     }
 
     /**
      * Flush all static state.
      */
     public static function flushState(): void
     {
-        static::$container = [];
+        static::$chains = [];
     }
 }