fix(grpc): resilient SwooleGrpcTransport — mutex, retry, reconnect and PHP 8.2 compat

src/Factory/Log/Exporter/OtlpGrpcLogExporterFactory.php

@@ -6,6 +6,7 @@
 
 use Hyperf\Contract\ConfigInterface;
 use Hyperf\OpenTelemetry\Transport\SwooleGrpcTransportFactory;
+use OpenTelemetry\Contrib\Otlp\ContentTypes;
 use OpenTelemetry\Contrib\Otlp\LogsExporter;
 use OpenTelemetry\SDK\Common\Export\TransportFactoryInterface;
 use OpenTelemetry\SDK\Logs\LogRecordExporterInterface;
@@ -28,7 +29,7 @@ public function make(): LogRecordExporterInterface
         return new LogsExporter(
             (new SwooleGrpcTransportFactory())->create(
                 endpoint: $endpoint,
-                contentType: 'application/x-protobuf',
+                contentType: ContentTypes::PROTOBUF,
                 headers: $options['headers'] ?? [],
                 compression: $options['compression'] ?? TransportFactoryInterface::COMPRESSION_GZIP,
                 timeout: $options['timeout'] ?? 10,

src/Factory/Metric/Exporter/OtlpGrpcMetricExporterFactory.php

@@ -6,6 +6,7 @@
 
 use Hyperf\Contract\ConfigInterface;
 use Hyperf\OpenTelemetry\Transport\SwooleGrpcTransportFactory;
+use OpenTelemetry\Contrib\Otlp\ContentTypes;
 use OpenTelemetry\Contrib\Otlp\MetricExporter;
 use OpenTelemetry\SDK\Common\Export\TransportFactoryInterface;
 use OpenTelemetry\SDK\Metrics\Data\Temporality;
@@ -24,12 +25,12 @@ public function make(): MetricExporterInterface
     {
         $options = $this->config->get('open-telemetry.metrics.exporters.otlp_grpc.options', []);
 
-        $endpoint = rtrim($options['endpoint'], '/') . self::GRPC_METHOD;
+        $endpoint = rtrim($options['endpoint'] ?? 'http://localhost:4317', '/') . self::GRPC_METHOD;
 
         return new MetricExporter(
             transport: (new SwooleGrpcTransportFactory())->create(
                 endpoint: $endpoint,
-                contentType: 'application/x-protobuf',
+                contentType: ContentTypes::PROTOBUF,
                 headers: $options['headers'] ?? [],
                 compression: $options['compression'] ?? TransportFactoryInterface::COMPRESSION_GZIP,
                 timeout: $options['timeout'] ?? 10,

src/Factory/Trace/Exporter/OtlpGrpcTraceExporterFactory.php

@@ -6,6 +6,7 @@
 
 use Hyperf\Contract\ConfigInterface;
 use Hyperf\OpenTelemetry\Transport\SwooleGrpcTransportFactory;
+use OpenTelemetry\Contrib\Otlp\ContentTypes;
 use OpenTelemetry\Contrib\Otlp\SpanExporter;
 use OpenTelemetry\SDK\Common\Export\TransportFactoryInterface;
 use OpenTelemetry\SDK\Trace\SpanExporterInterface;
@@ -23,12 +24,12 @@ public function make(): SpanExporterInterface
     {
         $options = $this->config->get('open-telemetry.traces.exporters.otlp_grpc.options', []);
 
-        $endpoint = rtrim($options['endpoint'], '/') . self::GRPC_METHOD;
+        $endpoint = rtrim($options['endpoint'] ?? 'http://localhost:4317', '/') . self::GRPC_METHOD;
 
         return new SpanExporter(
             (new SwooleGrpcTransportFactory())->create(
                 endpoint: $endpoint,
-                contentType: 'application/x-protobuf',
+                contentType: ContentTypes::PROTOBUF,
                 headers: $options['headers'] ?? [],
                 compression: $options['compression'] ?? TransportFactoryInterface::COMPRESSION_GZIP,
                 timeout: $options['timeout'] ?? 10,

src/Listener/OtelShutdownListener.php

@@ -4,6 +4,8 @@
 
 namespace Hyperf\OpenTelemetry\Listener;
 
+use Hyperf\Command\Event\AfterExecute;
+use Hyperf\Command\Event\AfterHandle;
 use Hyperf\Contract\StdoutLoggerInterface;
 use Hyperf\Coroutine\Coroutine;
 use Hyperf\Event\Contract\ListenerInterface;
@@ -25,6 +27,8 @@ public function listen(): array
     {
         return [
             OnWorkerExit::class,
+            AfterHandle::class,
+            AfterExecute::class,
         ];
     }
 

src/SpanProcessor/ChannelBatchSpanProcessor.php

@@ -49,19 +49,6 @@ public function __construct(
     ) {
     }
 
-    private function ensureInitialized(): void
-    {
-        if ($this->initialized) {
-            return;
-        }
-
-        $this->initialized = true;
-        $this->exportContext = Context::getCurrent();
-        $this->channel = new Channel($this->channelCapacity);
-        $this->startConsumer();
-        $this->startFlushTimer();
-    }
-
     public function onStart(ReadWriteSpanInterface $span, ContextInterface $parentContext): void
     {
     }
@@ -116,6 +103,19 @@ public function shutdown(?CancellationInterface $cancellation = null): bool
         return $this->exporter->shutdown($cancellation);
     }
 
+    private function ensureInitialized(): void
+    {
+        if ($this->initialized) {
+            return;
+        }
+
+        $this->initialized = true;
+        $this->exportContext = Context::getCurrent();
+        $this->channel = new Channel($this->channelCapacity);
+        $this->startConsumer();
+        $this->startFlushTimer();
+    }
+
     private function pushBatch(): void
     {
         if ($this->batch === [] || $this->channel === null) {

src/Transport/SwooleGrpcTransport.php

@@ -12,6 +12,8 @@
 use OpenTelemetry\SDK\Common\Future\ErrorFuture;
 use OpenTelemetry\SDK\Common\Future\FutureInterface;
 use RuntimeException;
+use Swoole\Coroutine;
+use Swoole\Coroutine\Channel;
 use Swoole\Coroutine\Http2\Client;
 use Swoole\Http2\Request;
 use Throwable;
@@ -22,6 +24,8 @@ final class SwooleGrpcTransport implements TransportInterface
 
     private ?Client $client = null;
 
+    private ?Channel $mutex = null;
+
     /**
      * @SuppressWarnings(PHPMD.BooleanArgumentFlag)
      */
@@ -33,6 +37,8 @@ public function __construct(
         private readonly float $timeout = 10.0,
         private readonly bool $ssl = false,
         private readonly ?string $compression = null,
+        private readonly int $retryDelay = 100,
+        private readonly int $maxRetries = 3,
     ) {
     }
 
@@ -47,42 +53,21 @@ public function send(string $payload, ?CancellationInterface $cancellation = nul
             return new ErrorFuture(new RuntimeException('Transport is closed'));
         }
 
-        try {
-            $client = $this->getClient();
-
-            $data = $this->compress($payload);
-
-            $request = new Request();
-            $request->method = 'POST';
-            $request->path = $this->method;
-            $request->headers = $this->buildHeaders();
-            $request->data = $this->packMessage($data);
-
-            $streamId = $client->send($request);
-
-            if ($streamId === false || $streamId <= 0) {
-                return new ErrorFuture(new RuntimeException(
-                    'Failed to send gRPC request: ' . ($client->errMsg ?: 'unknown error')
-                ));
-            }
-
-            $response = $client->recv($this->timeout);
-
-            if ($response === false) {
-                return new ErrorFuture(new RuntimeException(
-                    'Failed to receive gRPC response: ' . ($client->errMsg ?: 'timeout')
-                ));
-            }
+        $mutex = $this->getMutex();
+        if ($mutex->pop() === false) {
+            return new ErrorFuture(new RuntimeException('Transport is closed'));
+        }
 
-            $grpcStatus = $response->headers['grpc-status'] ?? '0';
-            if ($grpcStatus !== '0') {
-                $grpcMessage = $response->headers['grpc-message'] ?? 'Unknown error';
-                return new ErrorFuture(new RuntimeException("gRPC error: {$grpcMessage}", (int) $grpcStatus));
+        try {
+            if ($this->closed) {
+                return new ErrorFuture(new RuntimeException('Transport is closed'));
             }
 
-            return new CompletedFuture(null);
+            return $this->attemptSend($payload);
         } catch (Throwable $e) {
             return new ErrorFuture($e);
+        } finally {
+            $mutex->push(true); // release — returns false silently if channel was closed by shutdown
         }
     }
 
@@ -93,6 +78,7 @@ public function shutdown(?CancellationInterface $cancellation = null): bool
         }
 
         $this->closed = true;
+        $this->mutex?->close(); // unblock any coroutines waiting to acquire the mutex
 
         if ($this->client !== null) {
             $this->client->close();
@@ -107,19 +93,126 @@ public function forceFlush(?CancellationInterface $cancellation = null): bool
         return ! $this->closed;
     }
 
+    /**
+     * @SuppressWarnings(PHPMD.ErrorControlOperator)
+     */
+    private function attemptSend(string $payload): FutureInterface
+    {
+        $data = $this->compress($payload);
+        $lastError = null;
+
+        for ($attempt = 0; $attempt <= $this->maxRetries; ++$attempt) {
+            if ($this->closed) {
+                return new ErrorFuture(new RuntimeException('Transport is closed'));
+            }
+
+            if ($attempt > 0 && $this->retryDelay > 0) {
+                Coroutine::sleep($this->retryDelay / 1000.0);
+            }
+
+            try {
+                return $this->doRequest($this->getClient(), $data);
+            } catch (Throwable $e) {
+                $this->resetClient();
+                $lastError = $e;
+            }
+        }
+
+        return new ErrorFuture($lastError ?? new RuntimeException('Unknown transport error after retries'));
+    }
+
+    /**
+     * Executes a single gRPC send+recv cycle.
+     *
+     * Throws RuntimeException on send failure so the caller can reset the client and retry.
+     * Returns ErrorFuture on recv failure (delivery uncertain — no retry).
+     *
+     * Note: @$client->recv() suppresses E_DEPRECATED from Swoole setting $serverLastStreamId
+     * as a dynamic property in PHP 8.2+. Hyperf's ErrorExceptionHandler would otherwise convert
+     * the deprecation notice into an ErrorException, aborting the recv() call.
+     *
+     * @throws RuntimeException on retryable send failure
+     * @SuppressWarnings(PHPMD.ErrorControlOperator)
+     */
+    private function doRequest(Client $client, string $data): FutureInterface
+    {
+        $request = new Request();
+        $request->method = 'POST';
+        $request->path = $this->method;
+        $request->headers = $this->buildHeaders();
+        $request->data = $this->packMessage($data);
+
+        $streamId = $client->send($request);
+
+        if ($streamId === false || $streamId <= 0) {
+            // Throw so the retry loop can reset the client and try again;
+            // data was never transmitted so retrying is safe.
+            throw new RuntimeException(
+                'Failed to send gRPC request: ' . ($client->errMsg ?: 'unknown error')
+            );
+        }
+
+        $response = @$client->recv($this->timeout);
+
+        if ($response === false) {
+            $this->resetClient($client);
+            // Do not retry: send succeeded, so delivery is uncertain — retrying could duplicate exports.
+            return new ErrorFuture(new RuntimeException(
+                'Failed to receive gRPC response: ' . ($client->errMsg ?: 'timeout')
+            ));
+        }
+
+        $grpcStatus = (int) ($response->headers['grpc-status'] ?? 0);
+        if ($grpcStatus !== 0) {
+            $grpcMessage = $response->headers['grpc-message'] ?? 'Unknown error';
+            // gRPC application error — not retried
+            return new ErrorFuture(new RuntimeException("gRPC error: {$grpcMessage}", $grpcStatus));
+        }
+
+        return new CompletedFuture(null);
+    }
+
+    private function getMutex(): Channel
+    {
+        if ($this->mutex === null) {
+            $this->mutex = new Channel(1);
+            $this->mutex->push(true); // initially unlocked
+        }
+
+        return $this->mutex;
+    }
+
+    private function resetClient(?Client $client = null): void
+    {
+        $target = $client ?? $this->client;
+        $target?->close();
+        if ($target === $this->client) {
+            $this->client = null;
+        }
+    }
+
     private function getClient(): Client
     {
-        if ($this->client === null || ! $this->client->connected) {
-            $this->client = new Client($this->host, $this->port, $this->ssl);
-            $this->client->set([
+        if ($this->client !== null && ! $this->client->connected) {
+            $this->client->close();
+            $this->client = null;
+        }
+
+        if ($this->client === null) {
+            $client = new Client($this->host, $this->port, $this->ssl);
+            $client->set([
                 'timeout' => $this->timeout,
             ]);
 
-            if (! $this->client->connect()) {
+            if (! $client->connect()) {
+                $errMsg = $client->errMsg;
+                $client->close();
                 throw new RuntimeException(
-                    "Failed to connect to {$this->host}:{$this->port}: " . $this->client->errMsg
+                    "Failed to connect to {$this->host}:{$this->port}: " . $errMsg
                 );
             }
+
+            $this->client = $client;
         }
 
         return $this->client;
@@ -165,6 +258,7 @@ private function buildHeaders(): array
     private function packMessage(string $data): string
     {
         $compressed = $this->compression !== null ? 1 : 0;
+        // gRPC message frame: [1-byte compressed flag][4-byte big-endian message length][message]
         return pack('CN', $compressed, strlen($data)) . $data;
     }
 }

src/Transport/SwooleGrpcTransportFactory.php

@@ -11,6 +11,13 @@
 
 final class SwooleGrpcTransportFactory implements TransportFactoryInterface
 {
+    /**
+     * Note: $cacert, $cert, and $key are accepted to satisfy the TransportFactoryInterface
+     * contract but are not implemented in this transport. TLS certificates are not supported;
+     * use the https:// or grpcs:// scheme for basic TLS.
+     *
+     * @param null|mixed $compression
+     */
     public function create(
         string $endpoint,
         string $contentType = ContentTypes::PROTOBUF,
@@ -45,6 +52,13 @@ public function create(
             $compressionType = null;
         }
 
+        $supportedCompression = [null, TransportFactoryInterface::COMPRESSION_GZIP, TransportFactoryInterface::COMPRESSION_DEFLATE];
+        if (! in_array($compressionType, $supportedCompression, true)) {
+            throw new InvalidArgumentException(
+                sprintf('Unsupported compression type "%s"', $compressionType)
+            );
+        }
+
         return new SwooleGrpcTransport(
             host: $parsed['host'],
             port: $parsed['port'],
@@ -53,6 +67,8 @@ public function create(
             timeout: $timeout,
             ssl: $parsed['ssl'],
             compression: $compressionType,
+            retryDelay: $retryDelay,
+            maxRetries: $maxRetries,
         );
     }