Implement TCP keepalive

composer-dependency-analyser.php

@@ -6,4 +6,5 @@
 use ShipMonk\ComposerDependencyAnalyser\Config\ErrorType;
 
 return new Configuration()
-    ->ignoreErrorsOnExtension('ext-zlib', [ErrorType::SHADOW_DEPENDENCY]);
+    ->ignoreErrorsOnExtension('ext-zlib', [ErrorType::SHADOW_DEPENDENCY])
+    ->ignoreErrorsOnExtension('ext-sockets', [ErrorType::SHADOW_DEPENDENCY]);

composer.json

@@ -48,6 +48,9 @@
         "thesis/grpc-server": "*@dev",
         "thesis/protoregistry": "^0.1.2"
     },
+    "suggest": {
+        "ext-sockets": "Required for TCP keepalive (Client\\Builder::withKeepalive())."
+    },
     "repositories": [
         {
             "type": "path",

packages/client/composer.json

@@ -34,5 +34,8 @@
     "config": {
         "lock": false,
         "sort-packages": true
+    },
+    "suggest": {
+        "ext-sockets": "Required for TCP keepalive (Client\\Builder::withKeepalive())."
     }
 }

packages/client/src/Client/Builder.php

@@ -57,6 +57,8 @@ final class Builder
 
     private ?SocketConnector $connector = null;
 
+    private ?Internal\KeepaliveSettings $keepalive = null;
+
     private ?Encoder $encoder = null;
 
     /**
@@ -211,6 +213,29 @@ public function withSocketConnector(SocketConnector $connector): self
         return $builder;
     }
 
+    /**
+     * Enables TCP keepalive on every connection, so a dead peer is detected even on an idle
+     * connection after roughly `idle + interval * count` seconds. Applies on top of a custom
+     * {@see self::withSocketConnector()} connector, but not to a custom
+     * {@see self::withHttpClient()} client. Requires the "sockets" extension.
+     *
+     * @param positive-int $idle seconds of idleness before the first probe
+     * @param positive-int $interval seconds between unanswered probes
+     * @param positive-int $count unanswered probes before the connection is dropped
+     * @throws KeepaliveUnavailable
+     */
+    public function withKeepalive(int $idle = 10, int $interval = 10, int $count = 3): self
+    {
+        if (!\extension_loaded('sockets')) {
+            throw new KeepaliveUnavailable();
+        }
+
+        $builder = clone $this;
+        $builder->keepalive = new Internal\KeepaliveSettings($idle, $interval, $count);
+
+        return $builder;
+    }
+
     public function withLoadBalancer(LoadBalancerFactory $factory): self
     {
         $builder = clone $this;
@@ -272,11 +297,21 @@ public function build(): Client
             $controlMetadata,
         ]);
 
+        $connector = $this->connector ?? new DnsSocketConnector();
+        if ($this->keepalive !== null) {
+            $connector = new KeepaliveSocketConnector(
+                $connector,
+                $this->keepalive->idle,
+                $this->keepalive->interval,
+                $this->keepalive->count,
+            );
+        }
+
         $httpclient = $this->httpclient ?? new HttpClientBuilder()
             ->usingPool(ConnectionLimitingPool::byAuthority(
                 $this->connectionLimit,
                 new DefaultConnectionFactory(
-                    $this->connector ?? new DnsSocketConnector(),
+                    $connector,
                     new ConnectContext()
                         ->withConnectTimeout($this->connectTimeout)
                         ->withTlsContext($tlsContext),

packages/client/src/Client/Internal/KeepaliveSettings.php

@@ -0,0 +1,22 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Thesis\Grpc\Client\Internal;
+
+/**
+ * @internal
+ */
+final readonly class KeepaliveSettings
+{
+    /**
+     * @param positive-int $idle seconds of idleness before the first probe
+     * @param positive-int $interval seconds between unanswered probes
+     * @param positive-int $count unanswered probes before the connection is dropped
+     */
+    public function __construct(
+        public int $idle,
+        public int $interval,
+        public int $count,
+    ) {}
+}

packages/client/src/Client/KeepaliveSocketConnector.php

@@ -0,0 +1,105 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Thesis\Grpc\Client;
+
+use Amp\Cancellation;
+use Amp\Socket\ConnectContext;
+use Amp\Socket\ConnectException;
+use Amp\Socket\ResourceSocket;
+use Amp\Socket\Socket;
+use Amp\Socket\SocketAddress;
+use Amp\Socket\SocketConnector;
+
+/**
+ * Enables TCP keepalive on every connection made by the decorated connector, so that the
+ * operating system detects a dead peer (a crashed host, a network partition, a half-open
+ * connection) even while the connection is idle, e.g. a long-lived stream waiting for data.
+ *
+ * A dead peer is detected after roughly `idle + interval * count` seconds of silence.
+ * It does not detect a peer whose host is alive but whose process hangs.
+ *
+ * Requires the "sockets" extension.
+ *
+ * @api
+ */
+final readonly class KeepaliveSocketConnector implements SocketConnector
+{
+    /**
+     * @param positive-int $idle seconds of idleness before the first probe
+     * @param positive-int $interval seconds between unanswered probes
+     * @param positive-int $count unanswered probes before the connection is dropped
+     * @throws KeepaliveUnavailable
+     */
+    public function __construct(
+        private SocketConnector $connector,
+        private int $idle = 10,
+        private int $interval = 10,
+        private int $count = 3,
+    ) {
+        if (!\extension_loaded('sockets')) {
+            throw new KeepaliveUnavailable();
+        }
+    }
+
+    #[\Override]
+    public function connect(
+        SocketAddress|string $uri,
+        ?ConnectContext $context = null,
+        ?Cancellation $cancellation = null,
+    ): Socket {
+        $socket = $this->connector->connect($uri, $context, $cancellation);
+
+        if ($socket instanceof ResourceSocket) {
+            $this->enable($socket);
+        }
+
+        return $socket;
+    }
+
+    /**
+     * @throws ConnectException
+     */
+    private function enable(ResourceSocket $socket): void
+    {
+        $resource = $socket->getResource();
+        if (!\is_resource($resource)) {
+            return;
+        }
+
+        $raw = socket_import_stream($resource);
+        if ($raw === false) {
+            return;
+        }
+
+        $options = [[\SOL_SOCKET, \SO_KEEPALIVE, 1]];
+
+        // Linux and most BSDs, macOS names the idle time TCP_KEEPALIVE.
+        if (\defined('TCP_KEEPIDLE')) {
+            $options[] = [\SOL_TCP, \TCP_KEEPIDLE, $this->idle];
+        } elseif (\defined('TCP_KEEPALIVE') && \is_int($idle = \constant('TCP_KEEPALIVE'))) {
+            $options[] = [\SOL_TCP, $idle, $this->idle];
+        }
+
+        if (\defined('TCP_KEEPINTVL')) {
+            $options[] = [\SOL_TCP, \TCP_KEEPINTVL, $this->interval];
+        }
+
+        if (\defined('TCP_KEEPCNT')) {
+            $options[] = [\SOL_TCP, \TCP_KEEPCNT, $this->count];
+        }
+
+        foreach ($options as [$level, $option, $value]) {
+            if (!socket_set_option($raw, $level, $option, $value)) {
+                $socket->close();
+
+                throw new ConnectException(\sprintf(
+                    'Cannot enable TCP keepalive on the connection to "%s": %s',
+                    $socket->getRemoteAddress()->toString(),
+                    socket_strerror(socket_last_error($raw)),
+                ));
+            }
+        }
+    }
+}

packages/client/src/Client/KeepaliveUnavailable.php

@@ -0,0 +1,18 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Thesis\Grpc\Client;
+
+use Thesis\Grpc\GrpcException;
+
+/**
+ * @api
+ */
+final class KeepaliveUnavailable extends GrpcException
+{
+    public function __construct()
+    {
+        parent::__construct('TCP keepalive requires the "sockets" extension.');
+    }
+}

tests/Client/KeepaliveOptions.php

@@ -0,0 +1,33 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Thesis\Grpc\Client;
+
+use Amp\Socket\ResourceSocket;
+use PHPUnit\Framework\TestCase;
+
+/**
+ * Reads the TCP keepalive options of a connected socket (Linux option names).
+ */
+final class KeepaliveOptions
+{
+    /**
+     * @return array{keepalive: mixed, idle: mixed, interval: mixed, count: mixed}
+     */
+    public static function of(mixed $socket): array
+    {
+        TestCase::assertInstanceOf(ResourceSocket::class, $socket);
+        $resource = $socket->getResource();
+        TestCase::assertIsResource($resource);
+        $raw = socket_import_stream($resource);
+        TestCase::assertInstanceOf(\Socket::class, $raw);
+
+        return [
+            'keepalive' => socket_get_option($raw, \SOL_SOCKET, \SO_KEEPALIVE),
+            'idle' => socket_get_option($raw, \SOL_TCP, \TCP_KEEPIDLE),
+            'interval' => socket_get_option($raw, \SOL_TCP, \TCP_KEEPINTVL),
+            'count' => socket_get_option($raw, \SOL_TCP, \TCP_KEEPCNT),
+        ];
+    }
+}

tests/Client/KeepaliveSocketConnectorTest.php

@@ -0,0 +1,49 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Thesis\Grpc\Client;
+
+use Amp\Socket\DnsSocketConnector;
+use Amp\Socket\ServerSocket;
+use PHPUnit\Framework\Attributes\CoversClass;
+use PHPUnit\Framework\TestCase;
+use function Amp\Socket\listen;
+
+#[CoversClass(KeepaliveSocketConnector::class)]
+final class KeepaliveSocketConnectorTest extends TestCase
+{
+    private ServerSocket $server;
+
+    protected function setUp(): void
+    {
+        $this->server = listen('127.0.0.1:0');
+    }
+
+    protected function tearDown(): void
+    {
+        $this->server->close();
+    }
+
+    public function testEnablesKeepaliveWithTheGivenTimings(): void
+    {
+        $socket = new KeepaliveSocketConnector(new DnsSocketConnector(), idle: 7, interval: 3, count: 4)
+            ->connect($this->server->getAddress()->toString());
+
+        self::assertSame(
+            ['keepalive' => 1, 'idle' => 7, 'interval' => 3, 'count' => 4],
+            KeepaliveOptions::of($socket),
+        );
+
+        $socket->close();
+    }
+
+    public function testPlainConnectionsHaveNoKeepalive(): void
+    {
+        $socket = new DnsSocketConnector()->connect($this->server->getAddress()->toString());
+
+        self::assertSame(0, KeepaliveOptions::of($socket)['keepalive']);
+
+        $socket->close();
+    }
+}