fix(grpc): resilient SwooleGrpcTransport with mutex, retry and reconnect

lucas-angeli-gimenes · Copilot · lucas-angeli-gimenes · commit aee31a3b6531 · 2026-04-23T12:09:53.000-03:00
- Add Channel(1) mutex to serialize concurrent coroutine sends; shutdown
  closes the channel to unblock waiting coroutines without deadlock
- Extract attemptSend() retry loop (send failure → reset + retry up to
  maxRetries; recv failure → ErrorFuture immediately to avoid duplicate
  DELTA metric exports)
- Extract doRequest() to reduce cyclomatic complexity below threshold
- Reset HTTP/2 client on send/recv failure so next call reconnects
- Fix getClient(): close stale (disconnected) client before reconnecting
- Suppress PHP 8.2 E_DEPRECATED from Swoole's dynamic $serverLastStreamId
  property in recv() using @ operator (Hyperf ErrorExceptionHandler converts
  deprecation notices to ErrorException, aborting the recv)
- Validate unsupported compression types in factory (prevents compressed
  flag=1 with uncompressed payload)
- Wire retryDelay and maxRetries from factory to transport constructor
- Add endpoint fallback in Trace and Metric gRPC exporter factories
- Replace 'application/x-protobuf' literal with ContentTypes::PROTOBUF
- Cast grpc-status to int before comparison (more robust against whitespace)
- Add tests: default endpoint fallback, retry parameters, invalid compression

Co-authored-by: Copilot &lt;223556219+Copilot@users.noreply.github.com&gt;
diff --git a/src/Factory/Log/Exporter/OtlpGrpcLogExporterFactory.php b/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,
diff --git a/src/Factory/Metric/Exporter/OtlpGrpcMetricExporterFactory.php b/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,
diff --git a/src/Factory/Trace/Exporter/OtlpGrpcTraceExporterFactory.php b/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,
diff --git a/src/Transport/SwooleGrpcTransport.php b/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;
     }
 }
diff --git a/src/Transport/SwooleGrpcTransportFactory.php b/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,
         );
     }
 
diff --git a/tests/Unit/Factory/Metric/Exporter/OtlpGrpcMetricExporterFactoryTest.php b/tests/Unit/Factory/Metric/Exporter/OtlpGrpcMetricExporterFactoryTest.php
@@ -36,4 +36,17 @@ public function testMake(): void
 
         $this->assertInstanceOf(MetricExporterInterface::class, $exporter);
     }
+
+    public function testMakeWithDefaultEndpoint(): void
+    {
+        $config = $this->createMock(ConfigInterface::class);
+        $config->method('get')
+            ->with('open-telemetry.metrics.exporters.otlp_grpc.options', [])
+            ->willReturn([]);
+
+        $factory = new OtlpGrpcMetricExporterFactory($config);
+        $exporter = $factory->make();
+
+        $this->assertInstanceOf(MetricExporterInterface::class, $exporter);
+    }
 }
diff --git a/tests/Unit/Factory/Trace/Exporter/OtlpGrpcTraceExporterFactoryTest.php b/tests/Unit/Factory/Trace/Exporter/OtlpGrpcTraceExporterFactoryTest.php
@@ -34,4 +34,17 @@ public function testMake(): void
 
         $this->assertInstanceOf(SpanExporterInterface::class, $exporter);
     }
+
+    public function testMakeWithDefaultEndpoint(): void
+    {
+        $config = $this->createMock(ConfigInterface::class);
+        $config->method('get')
+            ->with('open-telemetry.traces.exporters.otlp_grpc.options', [])
+            ->willReturn([]);
+
+        $factory = new OtlpGrpcTraceExporterFactory($config);
+        $exporter = $factory->make();
+
+        $this->assertInstanceOf(SpanExporterInterface::class, $exporter);
+    }
 }
diff --git a/tests/Unit/Transport/SwooleGrpcTransportFactoryTest.php b/tests/Unit/Transport/SwooleGrpcTransportFactoryTest.php
@@ -119,4 +119,35 @@ public function testCreateWithCompressionNoneTreatedAsNull(): void
         $compressionProp = $reflection->getProperty('compression');
         $this->assertNull($compressionProp->getValue($transport));
     }
+
+    public function testCreateWithRetryParameters(): void
+    {
+        $factory = new SwooleGrpcTransportFactory();
+
+        $transport = $factory->create(
+            endpoint: 'http://localhost:4317/opentelemetry.proto.collector.trace.v1.TraceService/Export',
+            contentType: ContentTypes::PROTOBUF,
+            retryDelay: 200,
+            maxRetries: 5,
+        );
+
+        $reflection = new ReflectionClass($transport);
+
+        $this->assertSame(200, $reflection->getProperty('retryDelay')->getValue($transport));
+        $this->assertSame(5, $reflection->getProperty('maxRetries')->getValue($transport));
+    }
+
+    public function testCreateThrowsExceptionForUnsupportedCompression(): void
+    {
+        $factory = new SwooleGrpcTransportFactory();
+
+        $this->expectException(InvalidArgumentException::class);
+        $this->expectExceptionMessage('Unsupported compression type "br"');
+
+        $factory->create(
+            endpoint: 'http://localhost:4317/opentelemetry.proto.collector.trace.v1.TraceService/Export',
+            contentType: ContentTypes::PROTOBUF,
+            compression: 'br',
+        );
+    }
 }
