feat: report lame duck mode and let the application move off the server

README.md

@@ -123,6 +123,29 @@ The certificate is checked against `tlsCaFile`, not against the host name, for
 configured and discovered servers alike. Without `tlsCaFile` it is not checked at
 all.
 
+#### Lame duck mode
+
+A server being taken out of service announces it, then disconnects its clients
+gradually. `lameDuckModeHandler` reports the announcement; like the go, python and
+javascript clients, this one keeps the connection and leaves the decision to the
+application. Call `forceReconnect()` to move to another member, applied at the
+next read or write:
+
+```php
+$configuration = new Configuration(
+    servers: ['one.example.com:4222', 'two.example.com:4222'],
+    lameDuckModeHandler: fn (Client $client) => $client->forceReconnect(),
+);
+```
+
+`forceReconnect()` makes one attempt at each server and throws the connection
+error if none answers, rather than retrying until one does. That keeps the call
+that triggered it from blocking for ever when every member is draining or down.
+
+Doing nothing is also reasonable: the server closes the connection when it is
+ready, and the client then reconnects to another member on its own, having moved
+the draining server to the back of the queue.
+
 ### Connecting with TLS
 Typically, when connecting to a cluster with TLS enabled the connection settings do not change. The client lib will automatically switch over to TLS 1.2. However, if you're using a self-signed certificate you may have to point to your local CA file using the tlsCaFile setting.
 
@@ -503,6 +526,7 @@ The following is the list of configuration options and default values.
 | `ignoreDiscoveredServers` | `false` | Use only the configured servers, ignoring the members a cluster advertises.                                                                                                                 |
 | `inboxPrefix`          | `"_INBOX"` | Sets de prefix for automatically created inboxes                                                                                                                                            |
 | `jwt`                  |            | Token for [JWT Authentication](https://docs.nats.io/running-a-nats-service/configuration/securing_nats/auth_intro/jwt). Alternatively you can use [CredentialsParser](#connecting-with-jwt) |
+| `lameDuckModeHandler`  |            | Called when the connected server announces it is being taken out of service.                                                                                                                |
 | `maxReconnectAttempts` | `-1`       | Maximum failed connection attempts **per server** before that server is dropped from the pool. Negative = unlimited. Once every server is dropped, that recovery attempt gives up and raises the connection error; the next call starts again from the configured servers.                    |
 | `nkey`                 |            | Ed25519 based public key signature used for [NKEY Authentication](https://docs.nats.io/running-a-nats-service/configuration/securing_nats/auth_intro/nkey_auth).                            |
 | `noRandomize`          | `false`    | Connect to the servers in the order configured instead of shuffling them.                                                                                                                   |

src/Client.php

@@ -110,6 +110,18 @@ public function ping(): bool
         return $this->connection->ping();
     }
 
+    /**
+     * Moves the connection to another server in the pool, applied at the next read
+     * or write. This is how an application migrates off a server that announced
+     * lame duck mode, from inside the lameDuckModeHandler.
+     */
+    public function forceReconnect(): self
+    {
+        $this->connection->forceReconnect();
+
+        return $this;
+    }
+
     /** @return string[] every server the client may use, without credentials */
     public function getServers(): array
     {

src/Configuration.php

@@ -51,6 +51,7 @@ public function __construct(
         public float $reconnectWait = 0.2,
         public float $reconnectJitter = 0.1,
         public ?Closure $discoveredServersHandler = null,
+        public ?Closure $lameDuckModeHandler = null,
     ) {
 
         $this->setDelay($delay, $delayMode);

src/Connection.php

@@ -33,6 +33,9 @@ class Connection
     /** Guards the handshake against re-entering the reconnect loop. */
     private bool $connecting = false;
 
+    /** Set by forceReconnect(), acted on at the next read or write. */
+    private bool $reconnectRequested = false;
+
     /** True once any connection has been established, as opposed to attempted. */
     private bool $hasConnected = false;
 
@@ -122,6 +125,8 @@ public function getMessage(null|int|float $timeout = 0): ?Message
             $timeout = $this->config->timeout;
         }
 
+        $this->applyRequestedReconnect();
+
         $now = microtime(true);
         $max = $timeout > 0 ? $now + $timeout : PHP_FLOAT_MAX;
         $iteration = 0;
@@ -251,6 +256,9 @@ public function ping(): bool
 
     public function sendMessage(Message $message): void
     {
+        // Before init(), so that a request made while there is no connection is
+        // answered by the connection init() is about to make, not by a second one.
+        $this->applyRequestedReconnect();
         $this->init();
 
         while (true) {
@@ -653,6 +661,23 @@ private function processInfo(Info $info): void
             $notifications[] = fn () => $this->invokeHandler($this->config->discoveredServersHandler);
         }
 
+        if ($info->ldm ?? false) {
+            // Checked when delivered rather than now: a connection that is still being
+            // confirmed has not marked its server as connected yet, which clears the
+            // draining flag, and the flag is what makes this edge triggered.
+            $notifications[] = function () {
+                $server = $this->getPool()->current();
+
+                if ($server === null || $server->draining) {
+                    return;
+                }
+
+                $server->draining = true;
+                $this->logger?->info('server entered lame duck mode: ' . $server->getAddress());
+                $this->invokeHandler($this->config->lameDuckModeHandler);
+            };
+        }
+
         if ($this->connecting) {
             // A handler that publishes, as one asking to migrate might, would write
             // to a connection that has not sent CONNECT yet.
@@ -704,6 +729,55 @@ private function invokeHandler(?Closure $handler): void
         }
     }
 
+    /**
+     * Asks for the connection to be moved to another server.
+     *
+     * Deferred to the next read or write rather than done here, because the way to
+     * migrate off a draining server is to call this from the lame duck handler,
+     * which runs inside the read that delivered the notification. Reconnecting from
+     * there would tear down the call stack still processing that message.
+     *
+     * Applies whatever the reconnect option says, that one governing whether the
+     * client recovers on its own rather than whether it does as it is told. If no
+     * server can be reached, the failure raised is the connection error itself.
+     */
+    public function forceReconnect(): void
+    {
+        $this->reconnectRequested = true;
+    }
+
+    private function applyRequestedReconnect(): void
+    {
+        if (!$this->reconnectRequested || $this->connecting) {
+            return;
+        }
+
+        $this->reconnectRequested = false;
+
+        // Nothing to move off yet: the next connection picks a server regardless.
+        if (!is_resource($this->socket)) {
+            return;
+        }
+
+        $this->logger?->debug(
+            'reconnecting on request, leaving ' . $this->getPool()->current()?->getAddress()
+        );
+
+        // Deliberately not routed through processException(): this is a requested
+        // operation rather than a failure. Handing it a synthetic exception logged
+        // an error for an ordinary event, refused to run at all when automatic
+        // reconnection was disabled, and, once the pool was exhausted, reported
+        // that synthetic exception as the cause in place of the connection errors
+        // that actually stopped it.
+        //
+        // A single pass, however the budget is set. This is a call the application is
+        // waiting on, and with an unlimited budget sweeping until something answers
+        // would never return when nothing does. Recovering from a lost connection is
+        // the job of the failure path; here the answer to "move me" can be "no".
+        $this->getPool()->next();
+        $this->connect(retry: false);
+    }
+
     /**
      * Merged rather than replaced, because an asynchronous INFO carries only some of
      * the fields. Replacing would turn a property that callers read directly, such

tests/Functional/ClusterTest.php

@@ -136,7 +136,7 @@ public function testMessagesStillRouteAfterFailover(): void
 
         $before = $publisher->connection->getPool()->current()->getAddress();
 
-        $this->dropConnection($publisher);
+        $publisher->forceReconnect();
         $publisher->publish($subject, 'after failover');
 
         // The cluster routes the message whichever node the publisher is on, so
@@ -154,6 +154,41 @@ public function testMessagesStillRouteAfterFailover(): void
         $publisher->disconnect();
     }
 
+    /**
+     * Node three is signalled into lame duck mode, which makes it announce the
+     * transition and then drain its clients. It is restarted afterwards, so this
+     * test owns that node.
+     */
+    public function testLameDuckModeIsAnnouncedByTheServer(): void
+    {
+        if (!$this->dockerAvailable()) {
+            $this->markTestSkipped('docker compose is needed to signal a server into lame duck mode');
+        }
+
+        $notified = [];
+        $client = $this->clusterClient(
+            [self::NODES[3]['client']],
+            ['lameDuckModeHandler' => function (Client $client) use (&$notified) {
+                $notified[] = $client->connection->getPool()->current()->getAddress();
+            }]
+        );
+        $client->ping();
+
+        $this->signalLameDuckMode(3);
+
+        $threshold = microtime(true) + 10;
+        while ($notified === [] && microtime(true) < $threshold) {
+            try {
+                $client->process(0.2);
+            } catch (\Throwable) {
+                // The server drains the connection once it is in lame duck mode.
+                break;
+            }
+        }
+
+        $this->assertSame([self::NODES[3]['client']], $notified);
+    }
+
     public function tearDown(): void
     {
         // The inherited teardown deletes every stream on the default server, which
@@ -262,4 +297,29 @@ private function unavailable(string $reason): void
 
         $this->markTestSkipped($reason);
     }
+
+    private function dockerAvailable(): bool
+    {
+        exec('docker compose version 2>/dev/null', $output, $status);
+
+        return $status === 0;
+    }
+
+    private function signalLameDuckMode(int $node): void
+    {
+        $directory = escapeshellarg($this->getProjectRoot() . '/docker');
+        $service = escapeshellarg("nats-cluster-$node");
+
+        exec("docker compose --project-directory $directory kill -s SIGUSR2 $service 2>&1", $output, $status);
+
+        if ($status !== 0) {
+            $this->markTestSkipped('Could not signal the server: ' . implode("\n", $output));
+        }
+
+        // Restored for the rest of the suite: a server stays in lame duck mode until
+        // it is restarted, and stops accepting connections while it is.
+        register_shutdown_function(static function () use ($directory, $service) {
+            exec("docker compose --project-directory $directory restart $service 2>&1");
+        });
+    }
 }

tests/Functional/FailoverTest.php

@@ -144,6 +144,65 @@ public function testRejectedCredentialsRetireTheServer(): void
         $this->assertTrue($client->connection->getPool()->current()?->authenticationFailed ?? false);
     }
 
+    public function testForceReconnectEstablishesANewConnection(): void
+    {
+        $client = $this->createClient([
+            'servers' => [$this->live()],
+            'reconnectWait' => 0.0,
+            'reconnectJitter' => 0.0,
+        ]);
+        $client->ping();
+
+        $socket = new ReflectionProperty(Connection::class, 'socket');
+        $before = (int) $socket->getValue($client->connection);
+
+        $client->forceReconnect();
+
+        $this->assertTrue($client->ping());
+        $this->assertNotSame($before, (int) $socket->getValue($client->connection));
+    }
+
+    /**
+     * The documented way to leave a draining server: the lame duck handler asks for
+     * a reconnect, which is applied at the next read or write rather than unwinding
+     * the read that delivered the notification.
+     */
+    public function testLameDuckHandlerCanMigrateOffTheDrainingServer(): void
+    {
+        $client = $this->createClient([
+            'servers' => [$this->live()],
+            'reconnectWait' => 0.0,
+            'reconnectJitter' => 0.0,
+            'lameDuckModeHandler' => fn (Client $client) => $client->forceReconnect(),
+        ]);
+        $client->ping();
+
+        $socket = new ReflectionProperty(Connection::class, 'socket');
+        $before = (int) $socket->getValue($client->connection);
+
+        // Stand in for the draining server and announce it.
+        [$clientEnd, $serverEnd] = stream_socket_pair(STREAM_PF_UNIX, STREAM_SOCK_STREAM, 0);
+        fwrite($serverEnd, 'INFO {"ldm":true}' . "\r\n");
+        $socket->setValue($client->connection, $clientEnd);
+
+        $client->connection->getMessage(0);
+
+        $this->assertTrue(
+            $client->connection->getPool()->current()->draining,
+            'the server has to be recorded as draining'
+        );
+
+        // Applied here, on the next use of the connection.
+        $this->assertTrue($client->ping());
+        $this->assertNotSame($before, (int) $socket->getValue($client->connection));
+        $this->assertFalse(
+            $client->connection->getPool()->current()->draining,
+            'a fresh connection is not draining'
+        );
+
+        fclose($serverEnd);
+    }
+
     public function tearDown(): void
     {
         // The inherited teardown talks to the server, which these clients cannot.