Refuse messages over max_payload

README.md

@@ -355,6 +355,17 @@ $nc->onError(static function (\Throwable $e): void {
 });
 ```
 
+A fatal error makes the server close the connection: the subscriptions fail with it, it is passed to the error callback, and the next call connects again.
+A call interrupted by the close throws `Nats\Exception\ConnectionWasClosed` with the server's error as the previous exception when the server sent one:
+
+```php
+try {
+    $nc->publish('events.happens', new Nats\Message($payload));
+} catch (Nats\Exception\ConnectionWasClosed $e) {
+    dump($e->getPrevious()); // e.g. Nats\Exception\MaximumPayloadViolation for a payload over the limit of the user's account
+}
+```
+
 #### Validation
 
 Subjects and names are checked before anything is sent, following the rules of nats-server and nats.go:
@@ -368,6 +379,8 @@ Subjects and names are checked before anything is sent, following the rules of n
 - A key-value or object store bucket name matches `^[a-zA-Z0-9_-]+$`: `Nats\Exception\InvalidBucketName`.
 - A key-value key matches `^[-/_=.a-zA-Z0-9]+$` and is a valid subject; the keys to watch may contain wildcards: `Nats\Exception\InvalidKey`.
 - An object name is non-empty: `Nats\Exception\ObjectIsInvalid`.
+- A message (headers and payload together) is at most the `max_payload` the server announces, 1 MiB by default: `Nats\Exception\MaximumPayloadViolation`.
+  The server would close the connection on a larger one. An account limit lower than that is known only to the server, which then closes the connection.
 - A micro endpoint name matches `^[A-Za-z0-9_-]+$` (`Nats\Exception\InvalidName`), a group name is a subject without `>` (`Nats\Exception\InvalidSubject`).
 
 ## Nats JetStream

docker/nats/server.conf

@@ -20,6 +20,16 @@ accounts: {
       # Used by the tests of the NKey authentication, the seed is in tests/NkeyAuthTest.php.
       { nkey: UAANT3HVY2YJTAZ64BOYWU36YUE77A26VOBWQQTLFUYIUHMJBZCYXCGW }
     ]
+  },
+  # Used by the tests of the payload limits: INFO announces the server's max_payload (1 MiB), the account's lower
+  # limit applies only after CONNECT, so a message the client lets through is still rejected by the server.
+  LIMITED: {
+    users: [
+      { user: limited, password: Pswd1 }
+    ],
+    limits: {
+      max_payload: 1024
+    }
   }
 }
 

src/Client.php

@@ -142,12 +142,21 @@ public function rawPublish(
         ?Cancellation $cancellation = null,
     ): void {
         $connection = $this->connection($cancellation);
+        $info = $connection->info();
 
-        if ($message->headers !== null && \count($message->headers) > 0 && !$connection->info()->allowHeaders) {
-            throw Exception\FeatureIsNotSupported::forHeaders($connection->info()->serverVersion);
+        if ($message->headers !== null && \count($message->headers) > 0 && !$info->allowHeaders) {
+            throw Exception\FeatureIsNotSupported::forHeaders($info->serverVersion);
         }
 
-        $connection->execute(Internal\Command::pub($subject, $message, $replyTo), $cancellation);
+        $pub = Internal\Command::pub($subject, $message, $replyTo);
+
+        // The server closes the connection on a message over its limit, failing every subscription of the client:
+        // the message is refused before it is sent instead, the same way nats.go does it.
+        if ($info->maxPayload !== null && ($size = $pub->message->size()) > $info->maxPayload) {
+            throw Exception\MaximumPayloadViolation::forMessage($size, $info->maxPayload);
+        }
+
+        $connection->execute($pub, $cancellation);
     }
 
     /**

src/Exception/MaximumControlLineExceeded.php

@@ -11,8 +11,11 @@
  */
 final class MaximumControlLineExceeded extends NatsException
 {
-    public function __construct(int $code = 0, ?\Throwable $previous = null)
+    /**
+     * @param non-empty-string $message the server's error, e.g. "maximum control line exceeded" with the details
+     */
+    public function __construct(string $message = 'Maximum Control Line Exceeded')
     {
-        parent::__construct('Maximum Control Line Exceeded', $code, $previous);
+        parent::__construct($message);
     }
 }

src/Exception/MaximumPayloadViolation.php

@@ -11,8 +11,27 @@
  */
 final class MaximumPayloadViolation extends NatsException
 {
-    public function __construct(int $code = 0, ?\Throwable $previous = null)
+    /**
+     * The server received a message larger than its max_payload and closed the connection.
+     */
+    public static function fromServer(): self
     {
-        parent::__construct('Maximum Payload Violation', $code, $previous);
+        return new self('Maximum Payload Violation');
+    }
+
+    /**
+     * The client refused to publish a message larger than the max_payload the server announced, nothing was sent.
+     *
+     * @param positive-int $size
+     * @param positive-int $maxPayload
+     */
+    public static function forMessage(int $size, int $maxPayload): self
+    {
+        return new self("Maximum Payload Violation, the message of {$size} bytes exceeds the server limit of {$maxPayload} bytes");
+    }
+
+    private function __construct(string $message)
+    {
+        parent::__construct($message);
     }
 }

src/Internal/Connection/ConnectionInfo.php

@@ -17,15 +17,19 @@ public static function fromServerInfo(ServerInfo $info): self
             serverVersion: $info->version,
             allowHeaders: $info->headers,
             supportJetstream: $info->jetstream ?? false,
+            maxPayload: $info->maxPayload,
         );
     }
 
     /**
      * @param non-empty-string $serverVersion
+     * @param ?positive-int $maxPayload the largest message (headers and payload) the server accepts, in bytes,
+     *                                  null if the server does not announce it
      */
     public function __construct(
         public string $serverVersion,
         public bool $allowHeaders = false,
         public bool $supportJetstream = false,
+        public ?int $maxPayload = null,
     ) {}
 }

src/Internal/Connection/SocketConnection.php

@@ -4,6 +4,7 @@
 
 namespace Thesis\Nats\Internal\Connection;
 
+use Amp\ByteStream\StreamException;
 use Amp\Cancellation;
 use Amp\CancelledException;
 use Amp\DeferredFuture;
@@ -47,6 +48,13 @@ final class SocketConnection implements Connection
      */
     private ?DeferredFuture $handshake = null;
 
+    /**
+     * Completed once the reader stops, with the error the server sent right before closing the connection, if any.
+     *
+     * @var DeferredFuture<?\Throwable>
+     */
+    private readonly DeferredFuture $termination;
+
     /**
      * @param ?KeyPair $keypair the NKey of the user: the server gets its public key and the nonce signed with the secret one,
      *                          the seed itself never leaves the client
@@ -63,6 +71,10 @@ public function __construct(
         /** @var \SplQueue<DeferredFuture<Protocol\Frame>> $queue */
         $queue = new \SplQueue();
         $this->queue = $queue;
+
+        /** @var DeferredFuture<?\Throwable> $termination */
+        $termination = new DeferredFuture();
+        $this->termination = $termination;
     }
 
     /**
@@ -114,6 +126,10 @@ public function execute(Protocol\Frame|array $frames, ?Cancellation $cancellatio
                 $ack->ignore();
             }
 
+            if ($e instanceof StreamException) {
+                throw $this->closedWhileWriting($e);
+            }
+
             throw $e;
         }
 
@@ -187,6 +203,24 @@ private function handshake(Cancellation $cancellation): void
         }
     }
 
+    /**
+     * The server closes the socket right after sending a fatal error, often in the middle of a write: amp shuts
+     * the writing side down only, so the reader still gets the error, and the write fails with the server's reason
+     * rather than a bare stream error.
+     */
+    private function closedWhileWriting(StreamException $e): ConnectionWasClosed
+    {
+        try {
+            $reason = $this->termination->getFuture()->await(new TimeoutCancellation($this->config->connectionTimeout));
+        } catch (CancelledException) {
+            $reason = null;
+        }
+
+        return $reason !== null
+            ? new ConnectionWasClosed("The connection was closed by the server: {$reason->getMessage()}", previous: $reason)
+            : new ConnectionWasClosed("The connection was closed: {$e->getMessage()}", previous: $e);
+    }
+
     public function info(): ConnectionInfo
     {
         return $this->info ?? throw new ConnectionIsNotAvailable();
@@ -210,6 +244,7 @@ private function run(): void
         $queue = $this->queue;
         $hooks = $this->hooks;
         $socket = $this->socket;
+        $termination = $this->termination;
         $state = &$this->state;
         $handshake = &$this->handshake;
 
@@ -218,14 +253,20 @@ private function run(): void
             $queue,
             $hooks,
             $socket,
+            $termination,
             &$state,
             &$handshake,
         ): void {
+            // The server sends a fatal error right before closing the connection, so it is the last frame read.
+            $last = null;
+
             while ($state === ConnectionState::Alive) {
                 $error = null;
 
                 try {
                     while (($frame = $framer->readFrame()) !== null) {
+                        $last = $frame;
+
                         if ($frame instanceof Protocol\Ok || $frame instanceof Protocol\Err) {
                             if (!$queue->isEmpty()) {
                                 // Verbose mode: the acknowledgement belongs to the oldest command awaiting one.
@@ -282,6 +323,10 @@ private function run(): void
                     // The server closed the connection without answering the handshake.
                     $handshake?->error(new ConnectionIsNotAvailable());
                     $handshake = null;
+
+                    if (!$termination->isComplete()) {
+                        $termination->complete($last instanceof Protocol\Err ? $last->toException() : null);
+                    }
                 }
             }
         });

src/Internal/Protocol/Err.php

@@ -56,7 +56,9 @@ public function toException(): \Exception
             str_starts_with($this->message, 'Secure Connection - TLS Required') => new Exception\TLSRequired(),
             str_starts_with($this->message, 'Stale Connection') => new Exception\StaleConnection(),
             str_starts_with($this->message, 'Maximum Connections Exceeded') => new Exception\MaximumConnectionsExceeded(),
-            str_starts_with($this->message, 'Maximum Payload Violation') => new Exception\MaximumPayloadViolation(),
+            str_starts_with($this->message, 'Maximum Payload Violation') => Exception\MaximumPayloadViolation::fromServer(),
+            // Unlike the others, this one is sent lowercase, with the details of the violation after a colon.
+            str_starts_with(strtolower($this->message), 'maximum control line exceeded') => new Exception\MaximumControlLineExceeded($this->message),
             str_starts_with($this->message, 'Slow Consumer') => Exception\SlowConsumer::fromServer(),
             str_starts_with($this->message, 'Invalid Subject') => new Exception\InvalidSubject(),
             str_starts_with($this->message, 'Permissions Violation') => new Exception\PermissionViolated($this->message),

src/Internal/Protocol/Message.php

@@ -20,6 +20,16 @@ public function __construct(
         public ?Headers $headers = null,
     ) {}
 
+    /**
+     * The size of the message the server checks against max_payload: the encoded headers and the payload, in bytes.
+     *
+     * @return non-negative-int
+     */
+    public function size(): int
+    {
+        return \strlen($this->payload ?? '') + ($this->headers !== null ? \strlen(encodeHeaders($this->headers)) : 0);
+    }
+
     public function encode(): string
     {
         $buffer = '';

tests/Internal/Protocol/ErrTest.php

@@ -8,6 +8,7 @@
 use PHPUnit\Framework\Attributes\TestWith;
 use PHPUnit\Framework\TestCase;
 use Thesis\Nats\Exception\AuthorizationViolation;
+use Thesis\Nats\Exception\MaximumControlLineExceeded;
 use Thesis\Nats\Exception\MaximumPayloadViolation;
 use Thesis\Nats\Exception\PermissionViolated;
 
@@ -33,6 +34,10 @@ public function testToException(): void
         self::assertInstanceOf(AuthorizationViolation::class, new Err('Authorization Violation')->toException());
         self::assertInstanceOf(MaximumPayloadViolation::class, new Err('Maximum Payload Violation')->toException());
 
+        $e = new Err('maximum control line exceeded: State 4, max_control_line 4096, Buffer len 4100')->toException();
+        self::assertInstanceOf(MaximumControlLineExceeded::class, $e);
+        self::assertSame('maximum control line exceeded: State 4, max_control_line 4096, Buffer len 4100', $e->getMessage());
+
         $e = new Err('Permissions Violation for Publish to "events.happens"')->toException();
         self::assertInstanceOf(PermissionViolated::class, $e);
         self::assertSame('Permissions Violation for Publish to "events.happens"', $e->getMessage());