diff --git a/README.md b/README.md
index 888f585..b86bc24 100644
--- a/README.md
+++ b/README.md
@@ -7,6 +7,9 @@ Pure non-blocking (fiber based) strictly typed full-featured PHP driver for NATS
   - [Publish-Subscribe](#pub-sub)
   - [Queues](#queues)
   - [Request-Reply](#request-reply)
+  - [Request Many](#request-many)
+  - [Auto-unsubscribe](#auto-unsubscribe)
+  - [Slow consumers](#slow-consumers)
 - [NATS JetStream](https://docs.nats.io/nats-concepts/jetstream)
   - [Publish](#publish)
   - [Publish Options](#publish-options)
@@ -156,6 +159,162 @@ dump("Received response {$response->message->payload}");
 $nc->drain();
 ```
 
+A request waits for the reply until it is cancelled, unless a timeout is given: per call, or for all requests with `Nats\Config::$requestTimeout`
+(`?request_timeout=5` in the URI), which also applies to the JetStream API calls. A request without a reply in time throws `Nats\Exception\RequestTimedOut`:
+
+```php
+<?php
+
+declare(strict_types=1);
+
+require_once __DIR__ . '/vendor/autoload.php';
+
+use Thesis\Nats;
+
+$nc = new Nats\Client(Nats\Config::fromURI('tcp://127.0.0.1:4222?request_timeout=5'));
+
+try {
+    $response = $nc->request('foo.bar', new Nats\Message('Hello World!'), timeout: 0.5); // overrides the default
+} catch (Nats\Exception\RequestTimedOut $e) {
+    dump($e->getMessage());
+}
+```
+
+#### Request many
+
+`Client::requestMany()` publishes one request and yields the replies of all responders as they arrive (scatter-gather, [ADR-47](https://github.com/nats-io/nats-architecture-and-design/blob/main/adr/ADR-47.md)).
+The iterator ends after `maxMessages` replies, once `timeout` has passed since the request (`Config::$requestTimeout` by default), when no reply arrives within `stall`
+after the previous one, on the reply the `sentinel` matches, which is not yielded, or once the `cancellation` is requested (iterating then throws `Amp\CancelledException`).
+The `stall` is counted between the replies only, as in ADR-47, since the first reply may take longer to come: it does not end the wait for a silent responder.
+So at least one of the other conditions is required, otherwise `requestMany()` throws `\InvalidArgumentException` instead of possibly gathering forever.
+If nobody listens to the subject, iterating throws `Nats\Exception\RequestHasNoResponders`.
+
+```php
+<?php
+
+declare(strict_types=1);
+
+require_once __DIR__ . '/vendor/autoload.php';
+
+use Thesis\Nats;
+
+$nc = new Nats\Client(Nats\Config::default());
+
+foreach (['eu', 'us', 'asia'] as $region) {
+    $nc->subscribe('discovery', static function (Nats\Delivery $delivery) use ($region): void {
+        $delivery->reply(new Nats\Message($region));
+    });
+}
+
+// Whoever answers within a second.
+foreach ($nc->requestMany('discovery', timeout: 1) as $reply) {
+    dump("Found {$reply->message->payload}");
+}
+
+// The first two answers, and no more than 100ms of silence between them.
+$replies = $nc->requestMany('discovery', maxMessages: 2, stall: 0.1, timeout: 1);
+foreach ($replies as $reply) {
+    dump("Found {$reply->message->payload}");
+}
+
+$nc->subscribe('report', static function (Nats\Delivery $delivery) use ($nc): void {
+    if ($delivery->replyTo === null) {
+        return;
+    }
+
+    foreach (['line 1', 'line 2', 'line 3'] as $line) {
+        $nc->publish($delivery->replyTo, new Nats\Message($line));
+    }
+
+    $nc->publish($delivery->replyTo, new Nats\Message()); // end of stream
+});
+
+// A streamed reply, ended by an empty message.
+$lines = $nc->requestMany(
+    subject: 'report',
+    sentinel: static fn(Nats\Delivery $delivery): bool => ($delivery->message->payload ?? '') === '',
+    timeout: 5,
+);
+foreach ($lines as $line) {
+    dump($line->message->payload);
+}
+
+$nc->drain();
+```
+
+#### Auto-unsubscribe
+
+With `max`, the server removes the subscription after delivering that many messages. The subscription then completes once the handler has processed them:
+
+```php
+<?php
+
+declare(strict_types=1);
+
+require_once __DIR__ . '/vendor/autoload.php';
+
+use Thesis\Nats;
+
+$nc = new Nats\Client(Nats\Config::default());
+
+$subscription = $nc->subscribe('foo.bar', static function (Nats\Delivery $delivery): void {
+    dump("Received message {$delivery->message->payload}");
+}, max: 2);
+
+foreach (['1', '2', '3'] as $payload) {
+    $nc->publish('foo.bar', new Nats\Message($payload)); // the third one is not delivered
+}
+
+$subscription->awaitCompletion();
+
+// The same for an iterator: the loop ends after the third message.
+$deliveries = $nc->subscribeIterator('foo.baz', max: 3);
+
+$nc->drain();
+```
+
+#### Slow consumers
+
+The messages a subscription has received, but its handler has not taken yet, are pending. By default a subscription keeps at most 524288 pending messages
+and 64 MiB of their payload, like nats.go does; `pendingLimits` of `subscribe()` and `subscribeIterator()` changes that. The messages beyond the limits are dropped, and `Nats\Exception\SlowConsumer` with the subscription
+is passed to the error callback, once until the handler catches up.
+
+The `bufferSize` of `subscribe()` is a different limit: it is the capacity of the queue in front of the handler, and a message that does not fit
+is not dropped, but waits until the handler takes one. Such a waiting message is pending too, so the pending limits are reached only behind a full buffer.
+
+```php
+<?php
+
+declare(strict_types=1);
+
+require_once __DIR__ . '/vendor/autoload.php';
+
+use Thesis\Nats;
+
+$nc = new Nats\Client(Nats\Config::default());
+
+$nc->onError(static function (\Throwable $e): void {
+    if ($e instanceof Nats\Exception\SlowConsumer && $e->subscription !== null) {
+        $stats = $e->subscription->stats();
+        dump("{$e->subscription->subject}: {$stats->pendingMessages} pending, {$stats->dropped} dropped");
+    }
+});
+
+$subscription = $nc->subscribe(
+    subject: 'metrics.>',
+    handler: static function (Nats\Delivery $delivery): void {
+        // a slow handler
+    },
+    pendingLimits: new Nats\PendingLimits(messages: 10_000, bytes: null), // null for no limit
+);
+
+// The same for an iterator, here without any limits.
+$deliveries = $nc->subscribeIterator('events.>', pendingLimits: Nats\PendingLimits::unlimited());
+
+$stats = $subscription->stats();
+dump($stats->pendingMessages, $stats->pendingBytes, $stats->delivered, $stats->dropped);
+```
+
 #### Server errors
 
 The errors the server answers a particular call with are thrown from that call: `Nats\Exception\AuthorizationViolation` from the first call over a new connection,
@@ -175,7 +334,8 @@ $nc = new Nats\Client(Nats\Config::default());
 
 $nc->onError(static function (\Throwable $e): void {
     // e.g. Nats\Exception\PermissionViolated for a publish to a subject the user is not allowed to publish to,
-    // or Nats\Exception\SlowConsumer, after which the server closes the connection
+    // Nats\Exception\SlowConsumer, after which the server closes the connection,
+    // or Nats\Exception\SlowConsumer with a subscription that drops messages, see Slow consumers
     dump($e->getMessage());
 });
 ```
diff --git a/src/Client.php b/src/Client.php
index 5d2eed3..24e7371 100644
--- a/src/Client.php
+++ b/src/Client.php
@@ -6,10 +6,13 @@
 
 use Amp\Cancellation;
 use Amp\CancelledException;
+use Amp\CompositeCancellation;
 use Amp\Future;
 use Amp\Pipeline;
+use Amp\TimeoutCancellation;
 use Revolt\EventLoop;
 use Thesis\Nats\Exception\ConnectionWasClosed;
+use Thesis\Nats\Exception\RequestTimedOut;
 use Thesis\Nats\Internal\Connection;
 use Thesis\Nats\Internal\Hooks;
 use Thesis\Nats\Internal\Id;
@@ -154,6 +157,7 @@ public function rawPublish(
      * @param non-empty-string $subject
      * @param ?non-empty-string $queueGroup
      * @param positive-int $bufferSize
+     * @param ?positive-int $max the number of messages after which the subscription ends, see {@see subscribe()}
      * @return Iterator<Delivery>
      * @throws NatsException
      */
@@ -162,6 +166,8 @@ public function subscribeIterator(
         ?string $queueGroup = null,
         ?Cancellation $cancellation = null,
         int $bufferSize = self::DEFAULT_DELIVERY_BUFFER_SIZE,
+        ?int $max = null,
+        PendingLimits $pendingLimits = new PendingLimits(),
     ): Iterator {
         /** @var Pipeline\Queue<Delivery> $queue */
         $queue = new Pipeline\Queue(bufferSize: $bufferSize);
@@ -172,27 +178,11 @@ public function subscribeIterator(
             handler: static fn(mixed $delivery): bool => Iter\push($queue, $delivery),
             queueGroup: $queueGroup,
             cancellation: $cancellation,
+            max: $max,
+            pendingLimits: $pendingLimits,
         );
 
-        // The iterator ends the way the subscription does: cleanly after stop() or drain(), with the error
-        // after a failure, e.g. the connection closed by the server. Otherwise the consumer would wait forever.
-        async(static function () use ($queue, $subscription): void {
-            try {
-                $subscription->awaitCompletion();
-            } catch (\Throwable $e) {
-                if (!$queue->isComplete()) {
-                    $queue->error($e);
-                }
-
-                return;
-            }
-
-            if (!$queue->isComplete()) {
-                $queue->complete();
-            }
-        })->ignore();
-
-        return Iter\PipelineIterator::fromQueue($queue, $subscription);
+        return self::iterate($queue, $subscription);
     }
 
     /**
@@ -200,6 +190,9 @@ public function subscribeIterator(
      * @param callable(Delivery, Subscription): void $handler
      * @param ?non-empty-string $queueGroup
      * @param positive-int $bufferSize
+     * @param ?positive-int $max the number of messages after which the server removes the subscription (UNSUB <sid> <max>),
+     *                           the subscription is then drained and completes once the handler has processed them
+     * @param PendingLimits $pendingLimits the messages beyond them are dropped, unlike the ones beyond the buffer size
      * @throws NatsException
      * @throws \Throwable
      */
@@ -209,6 +202,8 @@ public function subscribe(
         ?string $queueGroup = null,
         int $bufferSize = 1_000,
         ?Cancellation $cancellation = null,
+        ?int $max = null,
+        PendingLimits $pendingLimits = new PendingLimits(),
     ): Subscription {
         Validate::subscribeSubject($subject);
 
@@ -216,47 +211,102 @@ public function subscribe(
             Validate::queueGroup($queueGroup);
         }
 
-        $subscriptionId = $this->subscriptionIdGenerator->nextId();
-        $unsubscribe = $this->unsubscribe(...);
-
-        $handler = new SubscriptionHandler(
-            unsubscribe: static function () use ($subscriptionId, $unsubscribe): void {
-                $unsubscribe($subscriptionId);
-            },
-            handler: $handler,
-            bufferSize: $bufferSize,
-        );
-
-        $this->subscribers[$subscriptionId] = new Subscriber(
-            subject: $subject,
-            queueGroup: $queueGroup,
-            push: $handler->push(...),
-            subscription: $subscription = $handler->subscription,
-        );
-
-        try {
-            $this->sub($subject, $subscriptionId, $queueGroup, $cancellation);
-        } catch (\Throwable $e) {
-            unset($this->subscribers[$subscriptionId]);
-
-            throw $e;
+        if ($max !== null) {
+            Validate::count($max, 'max');
         }
 
-        return $subscription;
+        return $this->register($subject, $handler, $queueGroup, $bufferSize, $cancellation, $max, $pendingLimits);
     }
 
     /**
      * @param non-empty-string $subject
+     * @param ?float $timeout in seconds, {@see Config::$requestTimeout} by default
+     * @throws RequestTimedOut if no reply arrives in time
      * @throws NatsException
      */
     public function request(
         string $subject,
         Message $message = new Message(),
         ?Cancellation $cancellation = null,
+        ?float $timeout = null,
     ): Delivery {
         Validate::publishSubject($subject);
 
-        return $this->rawRequest($subject, $message, $cancellation);
+        return $this->rawRequest($subject, $message, $cancellation, $timeout);
+    }
+
+    /**
+     * Scatter-gather, see ADR-47: publishes the request and yields the replies as they arrive. The iterator ends
+     * after $maxMessages replies, once $timeout has passed since the request, when no reply arrives within $stall
+     * after the previous one, on the reply $sentinel matches (which is not yielded), on $cancellation, or when stopped.
+     * The stall is counted between the replies only, as in ADR-47: the first one may take longer to come. So a silent
+     * responder is not ended by the stall, and at least one of the other conditions, besides stop(), is required.
+     *
+     * @param non-empty-string $subject
+     * @param ?positive-int $maxMessages
+     * @param ?float $stall in seconds
+     * @param ?\Closure(Delivery): bool $sentinel
+     * @param ?float $timeout in seconds, {@see Config::$requestTimeout} by default
+     * @param positive-int $bufferSize
+     * @return Iterator<Delivery> fails with {@see Exception\RequestHasNoResponders} if the first reply says nobody listens,
+     *                            and with {@see CancelledException} once $cancellation is requested
+     * @throws \InvalidArgumentException if nothing would end the gathering without a reply
+     * @throws NatsException
+     */
+    public function requestMany(
+        string $subject,
+        Message $message = new Message(),
+        ?int $maxMessages = null,
+        ?float $stall = null,
+        ?\Closure $sentinel = null,
+        ?float $timeout = null,
+        ?Cancellation $cancellation = null,
+        int $bufferSize = self::DEFAULT_DELIVERY_BUFFER_SIZE,
+    ): Iterator {
+        Validate::publishSubject($subject);
+
+        if ($maxMessages !== null) {
+            Validate::count($maxMessages, 'maxMessages');
+        }
+
+        if ($stall !== null) {
+            Validate::duration($stall, 'stall');
+        }
+
+        if ($timeout !== null) {
+            Validate::duration($timeout, 'timeout');
+        }
+
+        $timeout ??= $this->config->requestTimeout;
+        if ($maxMessages === null && $sentinel === null && $timeout === null && $cancellation === null) {
+            throw new \InvalidArgumentException(
+                'The gathering would never end without a reply: give maxMessages, sentinel, timeout or cancellation, or set Config::$requestTimeout. The stall is counted between the replies only.',
+            );
+        }
+
+        /** @var Pipeline\Queue<Delivery> $queue */
+        $queue = new Pipeline\Queue(bufferSize: $bufferSize);
+        $handler = new Rpc\RequestManyHandler($queue, $stall, $sentinel);
+
+        $subscription = $this->register(
+            subject: $inbox = Id\generateInboxId(),
+            handler: $handler,
+            bufferSize: $bufferSize,
+            cancellation: $cancellation,
+            max: $maxMessages,
+            onReceive: $handler->received(...),
+        );
+
+        try {
+            $handler->start($subscription, $timeout, $cancellation);
+            $this->publish($subject, $message, $inbox, $cancellation);
+        } catch (\Throwable $e) {
+            $subscription->stop();
+
+            throw $e;
+        }
+
+        return self::iterate($queue, $subscription);
     }
 
     /**
@@ -264,13 +314,20 @@ public function request(
      *
      * @internal
      * @param non-empty-string $subject
+     * @param ?float $timeout in seconds, {@see Config::$requestTimeout} by default
+     * @throws RequestTimedOut if no reply arrives in time
      * @throws NatsException
      */
     public function rawRequest(
         string $subject,
         Message $message = new Message(),
         ?Cancellation $cancellation = null,
+        ?float $timeout = null,
     ): Delivery {
+        if ($timeout !== null) {
+            Validate::duration($timeout, 'timeout');
+        }
+
         $subscribe = $this->subscribeCallback(...);
         $this->rpc ??= async(function () use ($subscribe): Rpc\Handler {
             $handler = new Rpc\Handler();
@@ -288,9 +345,28 @@ public function rawRequest(
             return $handler;
         });
 
-        return $this->rpc
-            ->await($cancellation)
-            ->request($subject, $message, $this, $cancellation);
+        $timeout ??= $this->config->requestTimeout;
+        if ($timeout === null) {
+            return $this->rpc
+                ->await($cancellation)
+                ->request($subject, $message, $this, $cancellation);
+        }
+
+        $deadline = new TimeoutCancellation($timeout);
+        $composite = $cancellation !== null ? new CompositeCancellation($cancellation, $deadline) : $deadline;
+
+        try {
+            return $this->rpc
+                ->await($composite)
+                ->request($subject, $message, $this, $composite);
+        } catch (CancelledException $e) {
+            // The caller's own cancellation wins, even if the deadline has passed as well.
+            if ($deadline->isRequested() && !($cancellation?->isRequested() ?? false)) {
+                throw RequestTimedOut::after($subject, $timeout, $e);
+            }
+
+            throw $e;
+        }
     }
 
     /**
@@ -358,6 +434,66 @@ public function __destruct()
         $connection?->close();
     }
 
+    /**
+     * Registers a validated subscription.
+     *
+     * @param non-empty-string $subject
+     * @param callable(Delivery, Subscription): void $handler
+     * @param ?non-empty-string $queueGroup
+     * @param positive-int $bufferSize
+     * @param ?positive-int $max
+     * @param ?\Closure(): void $onReceive called once a message arrives, before it waits for the handler
+     * @throws NatsException
+     * @throws \Throwable
+     */
+    private function register(
+        string $subject,
+        callable $handler,
+        ?string $queueGroup = null,
+        int $bufferSize = self::DEFAULT_DELIVERY_BUFFER_SIZE,
+        ?Cancellation $cancellation = null,
+        ?int $max = null,
+        PendingLimits $pendingLimits = new PendingLimits(),
+        ?\Closure $onReceive = null,
+    ): Subscription {
+        $subscriptionId = $this->subscriptionIdGenerator->nextId();
+        $unsubscribe = $this->unsubscribe(...);
+
+        $handler = new SubscriptionHandler(
+            subject: $subject,
+            queueGroup: $queueGroup,
+            unsubscribe: static function () use (
+                $subscriptionId,
+                $unsubscribe,
+            ): void {
+                $unsubscribe($subscriptionId);
+            },
+            handler: $handler,
+            bufferSize: $bufferSize,
+            onSlowConsumer: $this->reportError(...),
+            remaining: $max,
+            onReceive: $onReceive,
+            pendingLimits: $pendingLimits,
+        );
+
+        $this->subscribers[$subscriptionId] = new Subscriber(
+            subject: $subject,
+            queueGroup: $queueGroup,
+            push: $handler->push(...),
+            subscription: $subscription = $handler->subscription,
+        );
+
+        try {
+            $this->sub($subject, $subscriptionId, $queueGroup, $cancellation, $max);
+        } catch (\Throwable $e) {
+            unset($this->subscribers[$subscriptionId]);
+
+            throw $e;
+        }
+
+        return $subscription;
+    }
+
     /**
      * @param non-empty-string $subject
      * @param callable(Delivery): void $handler
@@ -399,18 +535,31 @@ private function subscribeCallback(
      * @param non-empty-string $subject
      * @param non-empty-string $sid
      * @param ?non-empty-string $queueGroup
+     * @param ?positive-int $max
      * @throws NatsException
      */
-    private function sub(string $subject, string $sid, ?string $queueGroup = null, ?Cancellation $cancellation = null): void
-    {
+    private function sub(
+        string $subject,
+        string $sid,
+        ?string $queueGroup = null,
+        ?Cancellation $cancellation = null,
+        ?int $max = null,
+    ): void {
         $connection = $this->connection($cancellation);
 
         try {
-            $connection->execute(Internal\Command::sub($subject, $sid, $queueGroup), $cancellation);
+            // Written at once, so that no message can slip past the limit between the two commands.
+            $connection->execute(
+                $max !== null
+                    ? [Internal\Command::sub($subject, $sid, $queueGroup), Internal\Command::unsub($sid, $max)]
+                    : Internal\Command::sub($subject, $sid, $queueGroup),
+                $cancellation,
+            );
         } catch (CancelledException $e) {
             // In verbose mode the SUB has already been sent when the wait for +OK is cancelled:
             // the server-side subscription must not outlive the failed call.
-            async($connection->execute(...), Internal\Command::unsub($sid))->ignore();
+            $future = async($connection->execute(...), Internal\Command::unsub($sid));
+            $future->ignore();
 
             throw $e;
         }
@@ -493,7 +642,8 @@ private function failSubscriptions(Protocol\Err $error, \Throwable $e): bool
 
         foreach ($this->subscribers as $subscriber) {
             if ($subscriber->subscription !== null && $subscriber->subject === $subject && $subscriber->queueGroup === $queueGroup) {
-                async($subscriber->subscription->error(...), $e)->ignore();
+                $future = async($subscriber->subscription->error(...), $e);
+                $future->ignore();
                 $failed = true;
             }
         }
@@ -578,7 +728,10 @@ private function closed(Connection\Connection $connection, ?\Throwable $error):
 
         // A pending setup has no requests to fail yet: its failure is thrown to the request that started it.
         if ($rpc !== null && $rpc->isComplete()) {
-            $completes[] = async(static function () use ($rpc, $e): void {
+            $completes[] = async(static function () use (
+                $rpc,
+                $e,
+            ): void {
                 self::awaitOrNull($rpc)?->shutdown(reason: $e);
             });
         }
@@ -596,6 +749,37 @@ private function closed(Connection\Connection $connection, ?\Throwable $error):
         }
     }
 
+    /**
+     * @param Pipeline\Queue<Delivery> $queue
+     * @return Iterator<Delivery>
+     */
+    private static function iterate(Pipeline\Queue $queue, Subscription $subscription): Iterator
+    {
+        // The iterator ends the way the subscription does: cleanly after stop() or drain(), with the error
+        // after a failure, e.g. the connection closed by the server. Otherwise the consumer would wait forever.
+        $future = async(static function () use (
+            $queue,
+            $subscription,
+        ): void {
+            try {
+                $subscription->awaitCompletion();
+            } catch (\Throwable $e) {
+                if (!$queue->isComplete()) {
+                    $queue->error($e);
+                }
+
+                return;
+            }
+
+            if (!$queue->isComplete()) {
+                $queue->complete();
+            }
+        });
+        $future->ignore();
+
+        return Iter\PipelineIterator::fromQueue($queue, $subscription);
+    }
+
     /**
      * Awaits the future, returning null if it has failed: the failure has already been reported
      * to the caller who initiated the operation, and there is nothing to clean up.
@@ -636,10 +820,16 @@ private function connection(?Cancellation $cancellation = null): Connection\Conn
             $reason = null;
 
             $connection->hooks()->onMessage($this->invokeSubscriber(...));
-            $connection->hooks()->onError(function (Hooks\ErrorReceived $event) use ($connection, &$reason): void {
+            $connection->hooks()->onError(function (Hooks\ErrorReceived $event) use (
+                $connection,
+                &$reason,
+            ): void {
                 $reason ??= $this->handleError($event, $connection);
             });
-            $connection->hooks()->onClose(function () use ($connection, &$reason): void {
+            $connection->hooks()->onClose(function () use (
+                $connection,
+                &$reason,
+            ): void {
                 $this->closed($connection, $reason);
             });
 
diff --git a/src/Config.php b/src/Config.php
index 3235810..84c2113 100644
--- a/src/Config.php
+++ b/src/Config.php
@@ -4,6 +4,8 @@
 
 namespace Thesis\Nats;
 
+use Thesis\Nats\Internal\Validation\Validate;
+
 /**
  * @api
  */
@@ -34,6 +36,8 @@
      * @param positive-int $maxPings the maximum number of pings that we have not received a response to, after which the connection to the server will be closed
      * @param ?non-empty-string $jetStreamDomain
      * @param non-empty-string $clientName
+     * @param ?float $requestTimeout in seconds, the default timeout of {@see Client::request()} and {@see Client::requestMany()}, null for none
+     * @throws \InvalidArgumentException
      */
     public function __construct(
         public array $urls = [self::DEFAULT_URL],
@@ -54,7 +58,12 @@ public function __construct(
         public int $maxPings = self::DEFAULT_MAX_PINGS,
         public ?string $jetStreamDomain = null,
         public string $clientName = self::DEFAULT_CLIENT_NAME,
+        public ?float $requestTimeout = null,
     ) {
+        if ($requestTimeout !== null) {
+            Validate::duration($requestTimeout, 'request timeout');
+        }
+
         $this->version = '0.1.x';
     }
 
@@ -162,6 +171,11 @@ public static function fromURI(#[\SensitiveParameter] string $uri): self
             $clientName = $query['client_name'];
         }
 
+        $requestTimeout = null;
+        if (isset($query['request_timeout']) && is_numeric($query['request_timeout']) && (float) $query['request_timeout'] > 0) {
+            $requestTimeout = (float) $query['request_timeout'];
+        }
+
         return new self(
             urls: $urls,
             verbose: $verbose,
@@ -177,6 +191,7 @@ public static function fromURI(#[\SensitiveParameter] string $uri): self
             maxPings: $maxPings,
             jetStreamDomain: $jetStreamDomain,
             clientName: $clientName,
+            requestTimeout: $requestTimeout,
         );
     }
 
@@ -196,7 +211,9 @@ public static function fromURI(#[\SensitiveParameter] string $uri): self
      *     max_pings?: positive-int,
      *     jetstream_domain?: non-empty-string,
      *     client_name?: non-empty-string,
+     *     request_timeout?: float,
      * } $options
+     * @throws \InvalidArgumentException
      */
     public static function fromArray(#[\SensitiveParameter] array $options): self
     {
@@ -215,6 +232,7 @@ public static function fromArray(#[\SensitiveParameter] array $options): self
             maxPings: $options['max_pings'] ?? self::DEFAULT_MAX_PINGS,
             jetStreamDomain: $options['jetstream_domain'] ?? null,
             clientName: $options['client_name'] ?? self::DEFAULT_CLIENT_NAME,
+            requestTimeout: $options['request_timeout'] ?? null,
         );
     }
 }
diff --git a/src/Exception/RequestTimedOut.php b/src/Exception/RequestTimedOut.php
new file mode 100644
index 0000000..ccf4a44
--- /dev/null
+++ b/src/Exception/RequestTimedOut.php
@@ -0,0 +1,21 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Thesis\Nats\Exception;
+
+use Thesis\Nats\NatsException;
+
+/**
+ * @api
+ */
+final class RequestTimedOut extends NatsException
+{
+    /**
+     * @param non-empty-string $subject
+     */
+    public static function after(string $subject, float $timeout, ?\Throwable $previous = null): self
+    {
+        return new self(\sprintf('No reply to the request on "%s" in %ss.', $subject, $timeout), previous: $previous);
+    }
+}
diff --git a/src/Exception/SlowConsumer.php b/src/Exception/SlowConsumer.php
index 70e2900..cd53409 100644
--- a/src/Exception/SlowConsumer.php
+++ b/src/Exception/SlowConsumer.php
@@ -5,14 +5,37 @@
 namespace Thesis\Nats\Exception;
 
 use Thesis\Nats\NatsException;
+use Thesis\Nats\PendingLimits;
+use Thesis\Nats\Subscription;
 
 /**
  * @api
  */
 final class SlowConsumer extends NatsException
 {
-    public function __construct(int $code = 0, ?\Throwable $previous = null)
+    /**
+     * The server drops the messages the client does not read fast enough, then closes the connection.
+     */
+    public static function fromServer(): self
     {
-        parent::__construct('Slow Consumer', $code, $previous);
+        return new self('Slow Consumer');
+    }
+
+    /**
+     * The subscription drops the messages beyond its pending limits, see {@see PendingLimits}.
+     */
+    public static function forSubscription(Subscription $subscription): self
+    {
+        return new self("Slow Consumer, messages dropped on \"{$subscription->subject}\"", $subscription);
+    }
+
+    /**
+     * @param ?Subscription $subscription the subscription that drops the messages, null if reported by the server
+     */
+    private function __construct(
+        string $message,
+        public readonly ?Subscription $subscription = null,
+    ) {
+        parent::__construct($message);
     }
 }
diff --git a/src/Internal/Connection/Connection.php b/src/Internal/Connection/Connection.php
index c2ae812..c038bf3 100644
--- a/src/Internal/Connection/Connection.php
+++ b/src/Internal/Connection/Connection.php
@@ -15,9 +15,12 @@
 interface Connection
 {
     /**
+     * Writes the frames at once, in verbose mode then waits for the acknowledgements of all of them.
+     *
+     * @param Protocol\Frame|non-empty-list<Protocol\Frame> $frames
      * @throws NatsException
      */
-    public function execute(Protocol\Frame $frame, ?Cancellation $cancellation = null): void;
+    public function execute(Protocol\Frame|array $frames, ?Cancellation $cancellation = null): void;
 
     public function hooks(): Hooks\Provider;
 
diff --git a/src/Internal/Connection/Framer.php b/src/Internal/Connection/Framer.php
index 53738f5..28fb4ad 100644
--- a/src/Internal/Connection/Framer.php
+++ b/src/Internal/Connection/Framer.php
@@ -29,7 +29,10 @@ public function __construct(Socket $socket)
         $queue = new Queue();
         $this->iterator = $queue->iterate();
 
-        EventLoop::queue(static function () use ($socket, $queue): void {
+        EventLoop::queue(static function () use (
+            $socket,
+            $queue,
+        ): void {
             $parser = new Protocol\Parser($queue->push(...));
 
             try {
diff --git a/src/Internal/Connection/PingPongHandler.php b/src/Internal/Connection/PingPongHandler.php
index b569265..bbc1041 100644
--- a/src/Internal/Connection/PingPongHandler.php
+++ b/src/Internal/Connection/PingPongHandler.php
@@ -4,15 +4,15 @@
 
 namespace Thesis\Nats\Internal\Connection;
 
-use Revolt\EventLoop;
 use Thesis\Nats\Internal\Protocol;
+use Thesis\Nats\Internal\Timer;
 
 /**
  * @internal
  */
 final class PingPongHandler
 {
-    private ?string $callbackId = null;
+    private ?Timer $timer = null;
 
     /** @var non-negative-int */
     private int $pings = 0;
@@ -29,7 +29,7 @@ public function startup(int $interval, int $maxPings): void
     {
         $interval /= 1_000;
 
-        $this->callbackId = EventLoop::repeat($interval, function () use ($maxPings): void {
+        $this->timer = Timer::repeat($interval, function () use ($maxPings): void {
             if (++$this->pings > $maxPings) {
                 $this->forceStop();
 
@@ -52,9 +52,7 @@ public function startup(int $interval, int $maxPings): void
 
     public function stop(): void
     {
-        if ($this->callbackId !== null) {
-            EventLoop::cancel($this->callbackId);
-        }
+        $this->timer?->cancel();
     }
 
     private function forceStop(): void
diff --git a/src/Internal/Connection/SocketConnection.php b/src/Internal/Connection/SocketConnection.php
index 5e3384a..4fc1372 100644
--- a/src/Internal/Connection/SocketConnection.php
+++ b/src/Internal/Connection/SocketConnection.php
@@ -7,6 +7,7 @@
 use Amp\Cancellation;
 use Amp\CancelledException;
 use Amp\DeferredFuture;
+use Amp\Future;
 use Amp\Socket\Socket;
 use Revolt\EventLoop;
 use Thesis\Nats\Config;
@@ -116,51 +117,49 @@ public function startup(): void
         }
     }
 
-    public function execute(Protocol\Frame $frame, ?Cancellation $cancellation = null): void
+    public function execute(Protocol\Frame|array $frames, ?Cancellation $cancellation = null): void
     {
-        /** @var ?DeferredFuture<Protocol\Frame> $deferred */
-        $deferred = null;
-
-        if ($this->config->verbose && self::isAcknowledged($frame)) {
-            /** @var DeferredFuture<Protocol\Frame> $deferred */
-            $deferred = new DeferredFuture();
-            $this->queue->push($deferred);
+        $frames = $frames instanceof Protocol\Frame ? [$frames] : $frames;
+
+        /** @var list<Future<Protocol\Frame>> $acks */
+        $acks = [];
+
+        if ($this->config->verbose) {
+            foreach ($frames as $frame) {
+                if (self::isAcknowledged($frame)) {
+                    /** @var DeferredFuture<Protocol\Frame> $deferred */
+                    $deferred = new DeferredFuture();
+                    $this->queue->push($deferred);
+                    $acks[] = $deferred->getFuture();
+                }
+            }
         }
 
         try {
-            $this->framer->writeFrame($frame);
+            $this->framer->writeFrame($frames);
         } catch (\Throwable $e) {
-            // The deferred is failed or dropped together with the rest of the queue once the connection is closed.
-            $deferred?->getFuture()->ignore();
+            // The deferreds are failed or dropped together with the rest of the queue once the connection is closed.
+            foreach ($acks as $ack) {
+                $ack->ignore();
+            }
 
             throw $e;
         }
 
-        if ($deferred === null) {
-            return;
-        }
-
-        $future = $deferred->getFuture();
+        foreach ($acks as $i => $ack) {
+            try {
+                $frame = $ack->await($cancellation);
+            } catch (CancelledException $e) {
+                $this->routeLateErrors(\array_slice($acks, $i));
 
-        try {
-            $frame = $future->await($cancellation);
-        } catch (CancelledException $e) {
-            // The deferred stays in the queue, so that the acknowledgement of this command is not attributed
-            // to the next one. An -ERR that arrives for it later is routed the same way as in non-verbose mode.
-            $hooks = $this->hooks;
-            $future
-                ->map(static function (Protocol\Frame $frame) use ($hooks): void {
-                    if ($frame instanceof Protocol\Err) {
-                        $hooks->dispatch(new Hooks\ErrorReceived($frame));
-                    }
-                })
-                ->ignore();
+                throw $e;
+            }
 
-            throw $e;
-        }
+            if ($frame instanceof Protocol\Err) {
+                $this->routeLateErrors(\array_slice($acks, $i + 1));
 
-        if ($frame instanceof Protocol\Err) {
-            throw $frame->toException();
+                throw $frame->toException();
+            }
         }
     }
 
@@ -286,6 +285,27 @@ private function connect(Protocol\Connect $connect): void
         }
     }
 
+    /**
+     * The deferreds of the acknowledgements nobody awaits anymore stay in the queue, so that they are not attributed
+     * to the next commands. An -ERR that arrives for them later is routed the same way as in non-verbose mode.
+     *
+     * @param list<Future<Protocol\Frame>> $acks
+     */
+    private function routeLateErrors(array $acks): void
+    {
+        $hooks = $this->hooks;
+
+        foreach ($acks as $ack) {
+            $ack
+                ->map(static function (Protocol\Frame $frame) use ($hooks): void {
+                    if ($frame instanceof Protocol\Err) {
+                        $hooks->dispatch(new Hooks\ErrorReceived($frame));
+                    }
+                })
+                ->ignore();
+        }
+    }
+
     /**
      * Whether the server answers the frame with +OK/-ERR in verbose mode. PING and PONG are the exception:
      * PING is answered with PONG and PONG with nothing, so waiting for their acknowledgement would consume
diff --git a/src/Internal/Protocol/Err.php b/src/Internal/Protocol/Err.php
index 08588c0..f8b8fde 100644
--- a/src/Internal/Protocol/Err.php
+++ b/src/Internal/Protocol/Err.php
@@ -57,7 +57,7 @@ public function toException(): \Exception
             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, 'Slow Consumer') => new Exception\SlowConsumer(),
+            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),
             default => new \RuntimeException($this->message),
diff --git a/src/Internal/Rpc/RequestManyHandler.php b/src/Internal/Rpc/RequestManyHandler.php
new file mode 100644
index 0000000..f60751c
--- /dev/null
+++ b/src/Internal/Rpc/RequestManyHandler.php
@@ -0,0 +1,140 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Thesis\Nats\Internal\Rpc;
+
+use Amp\Cancellation;
+use Amp\CancelledException;
+use Amp\Pipeline;
+use Thesis\Nats\Delivery;
+use Thesis\Nats\Exception\RequestHasNoResponders;
+use Thesis\Nats\Internal\Iter;
+use Thesis\Nats\Internal\Timer;
+use Thesis\Nats\Status;
+use Thesis\Nats\Subscription;
+use function Amp\async;
+
+/**
+ * Handles the inbox subscription of a scatter-gather request (ADR-47). The arrival of the replies, which the stall
+ * is counted from, is tracked apart from passing them to the consumer: the latter waits while the consumer is busy,
+ * and a slow consumer must not look like silent responders. The maximum number of replies is left to the subscription
+ * itself (UNSUB <sid> <max>).
+ *
+ * @internal
+ */
+final class RequestManyHandler
+{
+    private ?Subscription $subscription = null;
+
+    private bool $answered = false;
+
+    private ?Timer $stallTimer = null;
+
+    private ?Timer $timeoutTimer = null;
+
+    private ?string $cancellationId = null;
+
+    /**
+     * @param Pipeline\Queue<Delivery> $queue
+     * @param ?float $stall in seconds
+     * @param ?\Closure(Delivery): bool $sentinel
+     */
+    public function __construct(
+        private readonly Pipeline\Queue $queue,
+        private readonly ?float $stall = null,
+        private readonly ?\Closure $sentinel = null,
+    ) {}
+
+    /**
+     * Starts the clock once the inbox subscription exists, before the request is published.
+     *
+     * @param ?float $timeout in seconds
+     */
+    public function start(Subscription $subscription, ?float $timeout = null, ?Cancellation $cancellation = null): void
+    {
+        $this->subscription = $subscription;
+
+        // Whatever ends the gathering, the timers and the cancellation must not outlive it.
+        $future = async(function () use (
+            $subscription,
+            $cancellation,
+        ): void {
+            try {
+                $subscription->awaitCompletion();
+            } catch (\Throwable) {
+                // Reported to the consumer through the iterator.
+            } finally {
+                $this->stallTimer?->cancel();
+                $this->timeoutTimer?->cancel();
+
+                if ($this->cancellationId !== null) {
+                    $cancellation?->unsubscribe($this->cancellationId);
+                }
+            }
+        });
+        $future->ignore();
+
+        if ($timeout !== null) {
+            $this->timeoutTimer = Timer::delay($timeout, static fn() => self::end($subscription));
+        }
+
+        $this->cancellationId = $cancellation?->subscribe(static function (CancelledException $e) use ($subscription): void {
+            $future = async($subscription->error(...), $e);
+            $future->ignore();
+        });
+    }
+
+    /**
+     * Called once a reply arrives, however busy the consumer is: the stall is counted between the replies only.
+     */
+    public function received(): void
+    {
+        $this->stallTimer?->cancel();
+
+        $subscription = $this->subscription;
+
+        if ($this->stall !== null && $subscription !== null) {
+            $this->stallTimer = Timer::delay($this->stall, static fn() => self::end($subscription));
+        }
+    }
+
+    /**
+     * Passes the reply to the consumer, waiting while the consumer is busy.
+     *
+     * @throws RequestHasNoResponders if the first reply says nobody listens
+     */
+    public function __invoke(Delivery $delivery, Subscription $subscription): void
+    {
+        // Completed by the sentinel or by the consumer who stopped iterating: the rest of the replies are not needed.
+        if ($this->queue->isComplete()) {
+            return;
+        }
+
+        if (!$this->answered) {
+            $this->answered = true;
+
+            if ($delivery->message->headers?->statusCode() === Status::NoResponders) {
+                throw new RequestHasNoResponders();
+            }
+        }
+
+        if ($this->sentinel !== null && ($this->sentinel)($delivery)) {
+            $this->queue->complete();
+            $subscription->stop();
+
+            return;
+        }
+
+        Iter\push($this->queue, $delivery);
+    }
+
+    /**
+     * Ends the gathering on a timer: the replies already received are still yielded.
+     */
+    private static function end(Subscription $subscription): void
+    {
+        $future = async($subscription->drain(...));
+        $future->ignore();
+    }
+}
diff --git a/src/Internal/Subscription/Backlog.php b/src/Internal/Subscription/Backlog.php
new file mode 100644
index 0000000..e836304
--- /dev/null
+++ b/src/Internal/Subscription/Backlog.php
@@ -0,0 +1,112 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Thesis\Nats\Internal\Subscription;
+
+use Thesis\Nats\Delivery;
+use Thesis\Nats\PendingLimits;
+use Thesis\Nats\Subscription;
+use Thesis\Nats\SubscriptionStats;
+
+/**
+ * The messages a subscription has received, but not yet passed to its handler, shared by the handler, which accounts
+ * them, and the public {@see Subscription}, which exposes the stats. See {@see PendingLimits} for how the limits differ
+ * from the buffer size. Since every received message is dispatched in its own fiber, the limits also bound the number
+ * of fibers that pile up behind a slow handler.
+ *
+ * @internal
+ */
+final class Backlog
+{
+    /** @var non-negative-int */
+    private int $messages = 0;
+
+    /** @var non-negative-int */
+    private int $bytes = 0;
+
+    /** @var non-negative-int */
+    private int $delivered = 0;
+
+    /** @var non-negative-int */
+    private int $dropped = 0;
+
+    /** Whether the messages are being dropped: the overflow is reported once, until the handler catches up. */
+    private bool $overflowed = false;
+
+    /**
+     * @param \Closure(): void $onOverflow
+     */
+    public function __construct(
+        private readonly PendingLimits $limits,
+        private readonly \Closure $onOverflow,
+    ) {}
+
+    /**
+     * @return bool false if the message is dropped
+     */
+    public function admit(Delivery $delivery): bool
+    {
+        $size = self::size($delivery);
+
+        if (($this->limits->messages !== null && $this->messages >= $this->limits->messages)
+            || ($this->limits->bytes !== null && $this->bytes + $size > $this->limits->bytes)) {
+            ++$this->dropped;
+
+            if (!$this->overflowed) {
+                $this->overflowed = true;
+                ($this->onOverflow)();
+            }
+
+            return false;
+        }
+
+        $this->overflowed = false;
+        ++$this->messages;
+        $this->bytes += $size;
+
+        return true;
+    }
+
+    /**
+     * Forgets an admitted message that has not reached the handler, e.g. because the subscription has completed.
+     */
+    public function revoke(Delivery $delivery): void
+    {
+        $this->messages = max(0, $this->messages - 1);
+        $this->bytes = max(0, $this->bytes - self::size($delivery));
+    }
+
+    public function deliver(Delivery $delivery): void
+    {
+        $this->revoke($delivery);
+        ++$this->delivered;
+    }
+
+    public function clear(): void
+    {
+        $this->messages = 0;
+        $this->bytes = 0;
+    }
+
+    public function stats(): SubscriptionStats
+    {
+        return new SubscriptionStats(
+            pendingMessages: $this->messages,
+            pendingBytes: $this->bytes,
+            delivered: $this->delivered,
+            dropped: $this->dropped,
+        );
+    }
+
+    /**
+     * Only the payload is counted, as in nats.go: the limit is about the memory the application data takes,
+     * unlike {@see Delivery::size()}, which also counts the subject, the reply-to and the headers.
+     *
+     * @return non-negative-int
+     */
+    private static function size(Delivery $delivery): int
+    {
+        return \strlen($delivery->message->payload ?? '');
+    }
+}
diff --git a/src/Internal/Subscription/SubscriptionHandler.php b/src/Internal/Subscription/SubscriptionHandler.php
index 63c08fd..d723e72 100644
--- a/src/Internal/Subscription/SubscriptionHandler.php
+++ b/src/Internal/Subscription/SubscriptionHandler.php
@@ -9,6 +9,8 @@
 use Amp\Pipeline;
 use Revolt\EventLoop;
 use Thesis\Nats\Delivery;
+use Thesis\Nats\Exception\SlowConsumer;
+use Thesis\Nats\PendingLimits;
 use Thesis\Nats\Subscription;
 use function Amp\async;
 
@@ -22,18 +24,33 @@ final class SubscriptionHandler
     /** @var MessageQueue<Delivery> */
     private readonly MessageQueue $mq;
 
+    private readonly Backlog $backlog;
+
     private bool $inflight = false;
 
     /**
+     * @param non-empty-string $subject
+     * @param ?non-empty-string $queueGroup
      * @param \Closure(): void $unsubscribe
      * @param callable(Delivery, Subscription): void $handler
      * @param positive-int $bufferSize
+     * @param \Closure(SlowConsumer): void $onSlowConsumer
+     * @param ?positive-int $remaining the number of messages after which the server removes the subscription
+     * @param ?\Closure(): void $onReceive called once a message arrives, before it waits for the handler
      */
     public function __construct(
+        string $subject,
+        ?string $queueGroup,
         \Closure $unsubscribe,
         callable $handler,
         int $bufferSize,
+        private readonly \Closure $onSlowConsumer,
+        private ?int $remaining = null,
+        private readonly ?\Closure $onReceive = null,
+        PendingLimits $pendingLimits = new PendingLimits(),
     ) {
+        $this->backlog = $backlog = new Backlog($pendingLimits, $this->reportSlowConsumer(...));
+
         /** @var Pipeline\Queue<Delivery> $queue */
         $queue = new Pipeline\Queue($bufferSize);
 
@@ -45,6 +62,8 @@ public function __construct(
 
         $this->mq = $mq = new MessageQueue($queue);
         $this->subscription = $subscription = new Subscription(
+            $subject,
+            $queueGroup,
             $completeSubscriptionMarker->getFuture(),
             static function (Operation $op) use (
                 $unsubscribe,
@@ -53,6 +72,7 @@ static function (Operation $op) use (
                 $unsubscribe();
                 $completeSubscriptionDeferred->complete($op);
             },
+            $backlog,
         );
 
         $inflight = &$this->inflight;
@@ -65,6 +85,7 @@ static function (Operation $op) use (
             $subscription,
             $handler,
             $unsubscribe,
+            $backlog,
             &$inflight,
         ): void {
             while (!$queue->isComplete()) {
@@ -77,11 +98,13 @@ static function (Operation $op) use (
                 ]);
 
                 if ($op instanceof Emit) {
+                    /** @var Delivery $delivery */
+                    $delivery = $op->value;
+                    $backlog->deliver($delivery);
                     $inflight = true;
 
                     try {
-                        /** @phpstan-ignore argument.type */
-                        $handler($op->value, $subscription);
+                        $handler($delivery, $subscription);
                     } catch (\Throwable $e) {
                         try {
                             $unsubscribe();
@@ -90,6 +113,7 @@ static function (Operation $op) use (
                             // is the one to report, and it must not be lost in the event loop.
                         }
 
+                        $backlog->clear();
                         $completeSubscriptionMarker->error($e);
                         $queue->complete();
 
@@ -101,6 +125,7 @@ static function (Operation $op) use (
                     $queue->complete();
 
                     if ($op instanceof Error) {
+                        $backlog->clear();
                         $completeSubscriptionMarker->error($op->exception);
 
                         return;
@@ -119,9 +144,12 @@ static function (Operation $op) use (
                         }
 
                         foreach ($messages as $delivery) {
+                            $backlog->deliver($delivery);
+
                             try {
                                 $handler($delivery, $subscription);
                             } catch (\Throwable $e) {
+                                $backlog->clear();
                                 $completeSubscriptionMarker->error($e);
 
                                 return;
@@ -129,6 +157,7 @@ static function (Operation $op) use (
                         }
                     }
 
+                    $backlog->clear();
                     $completeSubscriptionMarker->complete();
                 }
             }
@@ -139,6 +168,24 @@ static function (Operation $op) use (
      * @return bool returns false if the queue has been closed, which should signal the client to remove the subscription
      */
     public function push(Delivery $delivery): bool
+    {
+        if ($this->onReceive !== null) {
+            ($this->onReceive)();
+        }
+
+        $last = $this->countDown();
+        $accepted = $this->accept($delivery);
+
+        if ($last && $accepted) {
+            // The server has removed the subscription after this message: the handler finishes the received ones.
+            $future = async($this->subscription->drain(...));
+            $future->ignore();
+        }
+
+        return $accepted;
+    }
+
+    private function accept(Delivery $delivery): bool
     {
         // If we are busy processing a message, we are unlikely to be interested in receiving heartbeats since we cannot process them.
         // To prevent accumulating them in memory and avoid triggering a flood of watchdog interactions after processing completes,
@@ -147,6 +194,42 @@ public function push(Delivery $delivery): bool
             return true;
         }
 
-        return $this->mq->push($delivery);
+        // Admitted before the push, which waits while the buffer is full: the waiting messages are pending as well.
+        if (!$this->backlog->admit($delivery)) {
+            return true;
+        }
+
+        if ($this->mq->push($delivery)) {
+            return true;
+        }
+
+        $this->backlog->revoke($delivery);
+
+        return false;
+    }
+
+    private function reportSlowConsumer(): void
+    {
+        ($this->onSlowConsumer)(SlowConsumer::forSubscription($this->subscription));
+    }
+
+    /**
+     * @return bool whether it is the last message the server delivers to the subscription
+     */
+    private function countDown(): bool
+    {
+        if ($this->remaining === null) {
+            return false;
+        }
+
+        if ($this->remaining > 1) {
+            --$this->remaining;
+
+            return false;
+        }
+
+        $this->remaining = null;
+
+        return true;
     }
 }
diff --git a/src/Internal/Timer.php b/src/Internal/Timer.php
new file mode 100644
index 0000000..85a019f
--- /dev/null
+++ b/src/Internal/Timer.php
@@ -0,0 +1,52 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Thesis\Nats\Internal;
+
+use Revolt\EventLoop;
+
+/**
+ * A scheduled callback of the event loop, cancelled explicitly or once the timer is no longer referenced,
+ * so the owner keeps it as long as the callback must run.
+ *
+ * @internal
+ */
+final readonly class Timer
+{
+    /**
+     * @param \Closure(): void $callback
+     */
+    public static function delay(float $seconds, \Closure $callback): self
+    {
+        return new self(EventLoop::delay($seconds, $callback));
+    }
+
+    /**
+     * @param \Closure(): void $callback
+     */
+    public static function repeat(float $interval, \Closure $callback): self
+    {
+        return new self(EventLoop::repeat($interval, $callback));
+    }
+
+    /**
+     * @param string $id the callback identifier of the event loop
+     */
+    private function __construct(
+        private string $id,
+    ) {}
+
+    /**
+     * Does nothing if the callback has already run or has been cancelled.
+     */
+    public function cancel(): void
+    {
+        EventLoop::cancel($this->id);
+    }
+
+    public function __destruct()
+    {
+        $this->cancel();
+    }
+}
diff --git a/src/Internal/Validation/Validate.php b/src/Internal/Validation/Validate.php
index 8f57779..fb0abe8 100644
--- a/src/Internal/Validation/Validate.php
+++ b/src/Internal/Validation/Validate.php
@@ -31,6 +31,36 @@ final class Validate
     /** nats-server validGroupName. */
     private const string PRIORITY_GROUP_REGEX = '/^[a-zA-Z0-9\/_=-]{1,16}$/';
 
+    /**
+     * A timeout, a stall or another duration in seconds.
+     *
+     * @param non-empty-string $what
+     * @throws \InvalidArgumentException
+     */
+    public static function duration(float $seconds, string $what): void
+    {
+        if (!is_finite($seconds) || $seconds <= 0) {
+            // NAN is not interpolated: PHP 8.5 warns when it is coerced to string.
+            $given = is_nan($seconds) ? 'NAN' : (string) $seconds;
+
+            throw new \InvalidArgumentException("The {$what} must be a positive finite number of seconds, {$given} given.");
+        }
+    }
+
+    /**
+     * A number of messages, which the phpdoc types promise to be positive: checked anyway, since zero
+     * would make the server remove a subscription the client keeps waiting on.
+     *
+     * @param non-empty-string $what
+     * @throws \InvalidArgumentException
+     */
+    public static function count(int $count, string $what): void
+    {
+        if ($count <= 0) {
+            throw new \InvalidArgumentException("The {$what} must be positive, {$count} given.");
+        }
+    }
+
     /**
      * A literal subject, e.g. of publish, request or reply-to: wildcards are not allowed.
      *
diff --git a/src/JetStream.php b/src/JetStream.php
index 9ecbd89..b935f5c 100644
--- a/src/JetStream.php
+++ b/src/JetStream.php
@@ -806,7 +806,10 @@ private function mapResponse(Message $message, string $type): object
      */
     private function publishPrefix(string $prefix, ?string $jsPrefix = null): \Closure
     {
-        return static function (string $cmd) use ($prefix, $jsPrefix): string {
+        return static function (string $cmd) use (
+            $prefix,
+            $jsPrefix,
+        ): string {
             $subject = '';
 
             if ($jsPrefix !== null) {
diff --git a/src/JetStream/Internal/Heartbeat/Watchdog.php b/src/JetStream/Internal/Heartbeat/Watchdog.php
index 25c4991..221de4d 100644
--- a/src/JetStream/Internal/Heartbeat/Watchdog.php
+++ b/src/JetStream/Internal/Heartbeat/Watchdog.php
@@ -4,8 +4,8 @@
 
 namespace Thesis\Nats\JetStream\Internal\Heartbeat;
 
-use Revolt\EventLoop;
 use Thesis\Nats\Exception\NoHeartbeatsReceived;
+use Thesis\Nats\Internal\Timer;
 use Thesis\Time\TimeSpan;
 
 /**
@@ -13,7 +13,7 @@
  */
 final class Watchdog
 {
-    private ?string $callbackId = null;
+    private ?Timer $timer = null;
 
     private readonly float $interval;
 
@@ -68,22 +68,14 @@ public function reset(): void
     public function stop(): void
     {
         $this->missed = 0;
-
-        if ($this->callbackId !== null) {
-            EventLoop::cancel($this->callbackId);
-            $this->callbackId = null;
-        }
-    }
-
-    public function __destruct()
-    {
-        $this->stop();
+        $this->timer?->cancel();
+        $this->timer = null;
     }
 
     private function schedule(): void
     {
         if ($this->interval > 0) {
-            $this->callbackId = EventLoop::repeat($this->interval, $this->func);
+            $this->timer = Timer::repeat($this->interval, $this->func);
         }
     }
 }
diff --git a/src/JetStream/PullConsumer.php b/src/JetStream/PullConsumer.php
index b4cfbfc..1331542 100644
--- a/src/JetStream/PullConsumer.php
+++ b/src/JetStream/PullConsumer.php
@@ -7,9 +7,9 @@
 use Amp\Cancellation;
 use Amp\Future;
 use Amp\Pipeline;
-use Revolt\EventLoop;
 use Thesis\Nats\Client;
 use Thesis\Nats\Internal\Id;
+use Thesis\Nats\Internal\Timer;
 use Thesis\Nats\JetStream\Api\PullRequest;
 use Thesis\Nats\JetStream\Api\Router;
 use Thesis\Nats\JetStream\Delivery as JetStreamDelivery;
@@ -84,14 +84,13 @@ public function fetch(FetchConfig $config, ?Cancellation $cancellation = null):
         );
 
         if ($config->noWait !== true) {
-            $callbackId = EventLoop::delay(
+            $timer = Timer::delay(
                 $config->maxWait->add(TimeSpan::fromSeconds(1))->toSeconds(),
                 static fn() => $subscription->stop(),
             );
 
-            $subscription = $subscription->onComplete(
-                static fn() => EventLoop::cancel($callbackId),
-            );
+            // Keeps the timer referenced, and so scheduled, until the subscription completes.
+            $subscription = $subscription->onComplete($timer->cancel(...));
         }
 
         $watchdog->subscribe($subscription->error(...));
diff --git a/src/Micro/Service.php b/src/Micro/Service.php
index 0a99911..21e8365 100644
--- a/src/Micro/Service.php
+++ b/src/Micro/Service.php
@@ -149,7 +149,10 @@ private function registerInternalHandler(
         $encoder = $this->encoder;
         $this->verbs[$name] = $this->nc->subscribe(
             $subject->value,
-            static function (Delivery $delivery) use ($handler, $encoder): void {
+            static function (Delivery $delivery) use (
+                $handler,
+                $encoder,
+            ): void {
                 $handler(new Request($delivery, $encoder));
             },
         );
diff --git a/src/PendingLimits.php b/src/PendingLimits.php
new file mode 100644
index 0000000..0cbe007
--- /dev/null
+++ b/src/PendingLimits.php
@@ -0,0 +1,48 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Thesis\Nats;
+
+use Thesis\Nats\Internal\Validation\Validate;
+
+/**
+ * Limits the messages a subscription has received, but its handler has not taken yet. The messages beyond the limits
+ * are dropped and reported to {@see Client::onError()} as {@see Exception\SlowConsumer}, once until the handler catches up.
+ * Only the payload is counted towards the bytes, as in nats.go.
+ *
+ * Unlike these limits, the buffer size of {@see Client::subscribe()} does not drop anything: it is the capacity
+ * of the queue in front of the handler, a message that does not fit waits until the handler takes one.
+ * Such a waiting message is pending as well, so the limits are reached only behind a full buffer.
+ *
+ * @api
+ */
+final readonly class PendingLimits
+{
+    /** The same defaults as in nats.go. */
+    public const int DEFAULT_MESSAGES = 512 * 1024;
+    public const int DEFAULT_BYTES = 64 * 1024 * 1024;
+
+    /**
+     * @param ?positive-int $messages null for no limit
+     * @param ?positive-int $bytes null for no limit
+     * @throws \InvalidArgumentException
+     */
+    public function __construct(
+        public ?int $messages = self::DEFAULT_MESSAGES,
+        public ?int $bytes = self::DEFAULT_BYTES,
+    ) {
+        if ($messages !== null) {
+            Validate::count($messages, 'pending messages limit');
+        }
+
+        if ($bytes !== null) {
+            Validate::count($bytes, 'pending bytes limit');
+        }
+    }
+
+    public static function unlimited(): self
+    {
+        return new self(messages: null, bytes: null);
+    }
+}
diff --git a/src/Subscription.php b/src/Subscription.php
index 7ddc17b..af6b314 100644
--- a/src/Subscription.php
+++ b/src/Subscription.php
@@ -6,6 +6,7 @@
 
 use Amp\Cancellation;
 use Amp\Future;
+use Thesis\Nats\Internal\Subscription\Backlog;
 use Thesis\Nats\Internal\Subscription\Drain;
 use Thesis\Nats\Internal\Subscription\Error;
 use Thesis\Nats\Internal\Subscription\Operation;
@@ -24,14 +25,25 @@ final class Subscription
     private array $onComplete = [];
 
     /**
+     * @internal
+     * @param non-empty-string $subject
+     * @param ?non-empty-string $queueGroup
      * @param Future<void> $completeMarker
      * @param \Closure(Operation<*>): void $complete
      */
     public function __construct(
+        public readonly string $subject,
+        public readonly ?string $queueGroup,
         private readonly Future $completeMarker,
         private readonly \Closure $complete,
+        private readonly Backlog $backlog,
     ) {}
 
+    public function stats(): SubscriptionStats
+    {
+        return $this->backlog->stats();
+    }
+
     /**
      * @param callable(): void ...$callbacks
      */
diff --git a/src/SubscriptionStats.php b/src/SubscriptionStats.php
new file mode 100644
index 0000000..4bfe031
--- /dev/null
+++ b/src/SubscriptionStats.php
@@ -0,0 +1,24 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Thesis\Nats;
+
+/**
+ * @api
+ */
+final readonly class SubscriptionStats
+{
+    /**
+     * @param non-negative-int $pendingMessages received, but not yet passed to the handler
+     * @param non-negative-int $pendingBytes the payload size of the pending messages
+     * @param non-negative-int $delivered passed to the handler
+     * @param non-negative-int $dropped dropped because the pending limits were exceeded
+     */
+    public function __construct(
+        public int $pendingMessages,
+        public int $pendingBytes,
+        public int $delivered,
+        public int $dropped,
+    ) {}
+}
diff --git a/tests/ClientTest.php b/tests/ClientTest.php
index 24124a8..4f229a6 100644
--- a/tests/ClientTest.php
+++ b/tests/ClientTest.php
@@ -373,12 +373,18 @@ private static function proxyToNats(string $listen, string $upstream): \Closure
             }
         };
 
-        EventLoop::queue(static function () use ($server, $upstream, $forward): void {
+        EventLoop::queue(static function () use (
+            $server,
+            $upstream,
+            $forward,
+        ): void {
             while (($client = $server->accept()) !== null) {
                 $nats = connect($upstream);
 
-                async($forward, $client, $nats)->ignore();
-                async($forward, $nats, $client)->ignore();
+                $future = async($forward, $client, $nats);
+                $future->ignore();
+                $future = async($forward, $nats, $client);
+                $future->ignore();
             }
         });
 
diff --git a/tests/ConfigTest.php b/tests/ConfigTest.php
index 5ebfe52..ef91f7a 100644
--- a/tests/ConfigTest.php
+++ b/tests/ConfigTest.php
@@ -183,6 +183,18 @@ final class ConfigTest extends TestCase
             ),
         ],
     )]
+    #[TestWith(
+        [
+            'tcp://127.0.0.1:4222?request_timeout=2.5',
+            new Config(requestTimeout: 2.5),
+        ],
+    )]
+    #[TestWith(
+        [
+            'tcp://127.0.0.1:4222?request_timeout=-1',
+            new Config(),
+        ],
+    )]
     public function testFromURI(string $uri, Config $config): void
     {
         self::assertEquals($config, Config::fromURI($uri));
@@ -221,4 +233,19 @@ public function testFromArrayWithoutJwtAndNkey(): void
 
         self::assertEquals($expected, $config);
     }
+
+    public function testFromArrayWithRequestTimeout(): void
+    {
+        self::assertEquals(new Config(requestTimeout: 0.5), Config::fromArray(['request_timeout' => 0.5]));
+    }
+
+    #[TestWith([0.0])]
+    #[TestWith([-1.0])]
+    #[TestWith([INF])]
+    #[TestWith([NAN])]
+    public function testRequestTimeoutMustBePositive(float $timeout): void
+    {
+        $this->expectException(\InvalidArgumentException::class);
+        new Config(requestTimeout: $timeout);
+    }
 }
diff --git a/tests/Internal/TimerTest.php b/tests/Internal/TimerTest.php
new file mode 100644
index 0000000..449f76f
--- /dev/null
+++ b/tests/Internal/TimerTest.php
@@ -0,0 +1,72 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Thesis\Nats\Internal;
+
+use PHPUnit\Framework\Attributes\CoversClass;
+use PHPUnit\Framework\TestCase;
+use function Amp\delay;
+
+#[CoversClass(Timer::class)]
+final class TimerTest extends TestCase
+{
+    public function testDelayRunsOnce(): void
+    {
+        /** @var \ArrayObject<int, true> $calls */
+        $calls = new \ArrayObject();
+        $timer = Timer::delay(0.01, static function () use ($calls): void {
+            $calls[] = true;
+        });
+
+        delay(0.05);
+
+        self::assertCount(1, $calls);
+        $timer->cancel(); // no-op once the callback has run
+    }
+
+    public function testCancelledDelayDoesNotRun(): void
+    {
+        /** @var \ArrayObject<int, true> $calls */
+        $calls = new \ArrayObject();
+        $timer = Timer::delay(0.01, static function () use ($calls): void {
+            $calls[] = true;
+        });
+
+        $timer->cancel();
+        delay(0.05);
+
+        self::assertCount(0, $calls);
+    }
+
+    public function testUnreferencedTimerIsCancelled(): void
+    {
+        /** @var \ArrayObject<int, true> $calls */
+        $calls = new \ArrayObject();
+        $timer = Timer::delay(0.01, static function () use ($calls): void {
+            $calls[] = true;
+        });
+
+        unset($timer);
+        delay(0.05);
+
+        self::assertCount(0, $calls);
+    }
+
+    public function testRepeatRunsUntilCancelled(): void
+    {
+        /** @var \ArrayObject<int, true> $calls */
+        $calls = new \ArrayObject();
+        $timer = Timer::repeat(0.01, static function () use ($calls): void {
+            $calls[] = true;
+        });
+
+        delay(0.055);
+        $timer->cancel();
+        $count = \count($calls);
+        delay(0.05);
+
+        self::assertGreaterThanOrEqual(3, $count);
+        self::assertCount($count, $calls);
+    }
+}
diff --git a/tests/JetStream/KeyValue/KeyValueTest.php b/tests/JetStream/KeyValue/KeyValueTest.php
index 2e83a33..511f056 100644
--- a/tests/JetStream/KeyValue/KeyValueTest.php
+++ b/tests/JetStream/KeyValue/KeyValueTest.php
@@ -329,7 +329,10 @@ public function testWatchHeadersOnly(): void
         $count = 0;
 
         $subscription = $kv->watch(
-            static function (Entry $entry, Subscription $subscription) use (&$changes, &$count): void {
+            static function (Entry $entry, Subscription $subscription) use (
+                &$changes,
+                &$count,
+            ): void {
                 $changes["{$entry->key}:{$entry->revision}"] = [$entry->value, $entry->size];
 
                 if (++$count >= 2) {
diff --git a/tests/JetStream/PullConsumerTest.php b/tests/JetStream/PullConsumerTest.php
index 3c0d162..fbe3539 100644
--- a/tests/JetStream/PullConsumerTest.php
+++ b/tests/JetStream/PullConsumerTest.php
@@ -351,7 +351,13 @@ public function testConsumePullsAgainAfterServerDroppedPullRequests(Status $stat
         // does when it drops waiting requests, e.g. on a consumer leader change.
         $observer = $nc->subscribe(
             subject: "\$JS.API.CONSUMER.MSG.NEXT.{$streamName}.{$consumerName}",
-            handler: static function (Delivery $delivery) use ($nc, $status, $description, &$pulls, $repulled): void {
+            handler: static function (Delivery $delivery) use (
+                $nc,
+                $status,
+                $description,
+                &$pulls,
+                $repulled,
+            ): void {
                 /** @var array{batch: int, ...} $pull */
                 $pull = json_decode($delivery->message->payload ?? '', true, flags: JSON_THROW_ON_ERROR);
 
diff --git a/tests/PausableProxy.php b/tests/PausableProxy.php
index ef0d1c9..01a9763 100644
--- a/tests/PausableProxy.php
+++ b/tests/PausableProxy.php
@@ -41,12 +41,18 @@ public static function start(string $upstream): self
 
         $proxy = new self($server, $address);
 
-        EventLoop::queue(static function () use ($proxy, $server, $upstream): void {
+        EventLoop::queue(static function () use (
+            $proxy,
+            $server,
+            $upstream,
+        ): void {
             while (($client = $server->accept()) !== null) {
                 $nats = connect($upstream);
 
-                async(static fn() => $proxy->forward($client, $nats, pausable: true))->ignore();
-                async(static fn() => $proxy->forward($nats, $client, pausable: false))->ignore();
+                $future = async(static fn() => $proxy->forward($client, $nats, pausable: true));
+                $future->ignore();
+                $future = async(static fn() => $proxy->forward($nats, $client, pausable: false));
+                $future->ignore();
             }
         });
 
diff --git a/tests/PendingLimitsTest.php b/tests/PendingLimitsTest.php
new file mode 100644
index 0000000..f93c103
--- /dev/null
+++ b/tests/PendingLimitsTest.php
@@ -0,0 +1,38 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Thesis\Nats;
+
+use PHPUnit\Framework\Attributes\CoversClass;
+use PHPUnit\Framework\Attributes\TestWith;
+use PHPUnit\Framework\TestCase;
+
+#[CoversClass(PendingLimits::class)]
+final class PendingLimitsTest extends TestCase
+{
+    public function testDefaultsAreTheOnesOfNatsGo(): void
+    {
+        $limits = new PendingLimits();
+
+        self::assertSame(512 * 1024, $limits->messages);
+        self::assertSame(64 * 1024 * 1024, $limits->bytes);
+    }
+
+    public function testUnlimited(): void
+    {
+        self::assertEquals(new PendingLimits(messages: null, bytes: null), PendingLimits::unlimited());
+    }
+
+    #[TestWith([0, null])]
+    #[TestWith([-1, null])]
+    #[TestWith([null, 0])]
+    #[TestWith([null, -1])]
+    public function testLimitsMustBePositive(?int $messages, ?int $bytes): void
+    {
+        $this->expectException(\InvalidArgumentException::class);
+
+        /** @phpstan-ignore argument.type, argument.type */
+        new PendingLimits($messages, $bytes);
+    }
+}
diff --git a/tests/RequestTest.php b/tests/RequestTest.php
new file mode 100644
index 0000000..ea69259
--- /dev/null
+++ b/tests/RequestTest.php
@@ -0,0 +1,380 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Thesis\Nats;
+
+use Amp\CancelledException;
+use Amp\DeferredCancellation;
+use Amp\TimeoutCancellation;
+use PHPUnit\Framework\Attributes\CoversClass;
+use PHPUnit\Framework\Attributes\TestWith;
+use Thesis\Nats\Exception\RequestHasNoResponders;
+use Thesis\Nats\Exception\RequestTimedOut;
+use Thesis\Nats\Internal\Id;
+use function Amp\async;
+use function Amp\delay;
+
+#[CoversClass(Client::class)]
+final class RequestTest extends NatsTestCase
+{
+    public function testRequestTimesOut(): void
+    {
+        $client = $this->client();
+
+        $id = Id\generateUniqueId();
+        $client->subscribe("{$id}.ignored", static fn() => null);
+
+        $started = microtime(true);
+
+        try {
+            $client->request("{$id}.ignored", new Message('Are you ok?'), timeout: 0.2);
+            self::fail('Nobody answers the request, it must time out.');
+        } catch (RequestTimedOut $e) {
+            self::assertInstanceOf(CancelledException::class, $e->getPrevious());
+        }
+
+        self::assertGreaterThanOrEqual(0.19, microtime(true) - $started, 'Timers may fire a bit early.');
+        self::assertSame(0, self::pendingRequests($client));
+    }
+
+    public function testRequestTimesOutByConfigDefault(): void
+    {
+        $config = $this->config();
+        $client = $this->clientWithConfig(new Config(
+            urls: $config->urls,
+            user: $config->user,
+            password: $config->password,
+            noResponders: $config->noResponders,
+            requestTimeout: 0.2,
+        ));
+
+        $id = Id\generateUniqueId();
+        $client->subscribe("{$id}.ignored", static fn() => null);
+        $client->subscribe("{$id}.answered", static function (Delivery $delivery): void {
+            $delivery->reply(new Message('ok'));
+        });
+
+        self::assertSame('ok', $client->request("{$id}.answered")->message->payload);
+
+        $this->expectException(RequestTimedOut::class);
+        $client->request("{$id}.ignored");
+    }
+
+    public function testExplicitTimeoutOverridesConfigDefault(): void
+    {
+        $config = $this->config();
+        $client = $this->clientWithConfig(new Config(
+            urls: $config->urls,
+            user: $config->user,
+            password: $config->password,
+            noResponders: $config->noResponders,
+            requestTimeout: 0.1,
+        ));
+
+        $id = Id\generateUniqueId();
+        $client->subscribe("{$id}.slow", static function (Delivery $delivery): void {
+            delay(0.3);
+            $delivery->reply(new Message('ok'));
+        });
+
+        self::assertSame('ok', $client->request("{$id}.slow", timeout: 2)->message->payload);
+    }
+
+    public function testCallerCancellationIsNotReportedAsTimeout(): void
+    {
+        $client = $this->client();
+
+        $id = Id\generateUniqueId();
+        $client->subscribe("{$id}.ignored", static fn() => null);
+
+        $this->expectException(CancelledException::class);
+        $client->request("{$id}.ignored", cancellation: new TimeoutCancellation(0.1), timeout: 5);
+    }
+
+    public function testRequestManyStopsAfterMaxMessages(): void
+    {
+        $client = $this->client();
+
+        $id = Id\generateUniqueId();
+        self::respond($client, "{$id}.scatter", ['a', 'b', 'c']);
+
+        $replies = self::payloads($client->requestMany("{$id}.scatter", new Message('?'), maxMessages: 2, timeout: 5));
+
+        self::assertCount(2, $replies);
+    }
+
+    public function testRequestManyGathersUntilTimeout(): void
+    {
+        $client = $this->client();
+
+        $id = Id\generateUniqueId();
+        self::respond($client, "{$id}.scatter", ['a', 'b', 'c']);
+
+        $started = microtime(true);
+        $replies = self::payloads($client->requestMany("{$id}.scatter", new Message('?'), timeout: 0.3));
+
+        self::assertGreaterThanOrEqual(0.29, microtime(true) - $started, 'Timers may fire a bit early.');
+        self::assertEqualsCanonicalizing(['a', 'b', 'c'], $replies);
+    }
+
+    public function testRequestManyStopsWhenRepliesStall(): void
+    {
+        $client = $this->client();
+
+        $id = Id\generateUniqueId();
+        $client->subscribe("{$id}.scatter", static function (Delivery $delivery) use ($client): void {
+            self::assertNotNull($delivery->replyTo);
+            $client->publish($delivery->replyTo, new Message('1'));
+            $client->publish($delivery->replyTo, new Message('2'));
+            delay(0.5);
+            $client->publish($delivery->replyTo, new Message('late'));
+        });
+
+        $started = microtime(true);
+        $replies = self::payloads($client->requestMany("{$id}.scatter", new Message('?'), stall: 0.2, timeout: 5));
+
+        self::assertSame(['1', '2'], $replies);
+        self::assertLessThan(0.5, microtime(true) - $started);
+    }
+
+    public function testSlowConsumerIsNotMistakenForStall(): void
+    {
+        $client = $this->client();
+
+        $id = Id\generateUniqueId();
+        $client->subscribe("{$id}.scatter", static function (Delivery $delivery) use ($client): void {
+            self::assertNotNull($delivery->replyTo);
+
+            for ($i = 0; $i < 10; ++$i) {
+                $client->publish($delivery->replyTo, new Message((string) $i));
+                delay(0.025);
+            }
+        });
+
+        // The replies come more often than the stall, but the consumer takes them slower than they come:
+        // they pile up in front of the full buffer, which is not a stall.
+        $replies = $client->requestMany("{$id}.scatter", new Message('?'), stall: 0.12, timeout: 10, bufferSize: 1);
+
+        $received = [];
+        foreach ($replies as $reply) {
+            $received[] = $reply->message->payload;
+            delay(0.2);
+        }
+
+        self::assertSame(array_map(strval(...), range(0, 9)), $received);
+    }
+
+    public function testRequestManyStopsOnSentinel(): void
+    {
+        $client = $this->client();
+
+        $id = Id\generateUniqueId();
+        $client->subscribe("{$id}.stream", static function (Delivery $delivery) use ($client): void {
+            self::assertNotNull($delivery->replyTo);
+            $client->publish($delivery->replyTo, new Message('1'));
+            $client->publish($delivery->replyTo, new Message('2'));
+            $client->publish($delivery->replyTo, new Message());
+            $client->publish($delivery->replyTo, new Message('after the end'));
+        });
+
+        $replies = self::payloads($client->requestMany(
+            "{$id}.stream",
+            new Message('?'),
+            sentinel: static fn(Delivery $delivery): bool => ($delivery->message->payload ?? '') === '',
+            timeout: 5,
+        ));
+
+        self::assertSame(['1', '2'], $replies);
+    }
+
+    public function testRequestManyFailsWithoutResponders(): void
+    {
+        $client = $this->client();
+
+        $this->expectException(RequestHasNoResponders::class);
+        self::payloads($client->requestMany(Id\generateUniqueId() . '.nobody', new Message('?'), timeout: 5));
+    }
+
+    public function testRequestManyFailsOnCancellation(): void
+    {
+        $client = $this->client();
+
+        $id = Id\generateUniqueId();
+        self::respond($client, "{$id}.scatter", ['a']);
+
+        $cancellation = new DeferredCancellation();
+        $replies = $client->requestMany("{$id}.scatter", new Message('?'), cancellation: $cancellation->getCancellation());
+
+        $received = [];
+
+        try {
+            foreach ($replies as $reply) {
+                $received[] = $reply->message->payload;
+                $cancellation->cancel();
+            }
+
+            self::fail('The gathering must fail once cancelled.');
+        } catch (CancelledException) {
+        }
+
+        self::assertSame(['a'], $received);
+    }
+
+    public function testStoppedRequestManyUnsubscribesFromInbox(): void
+    {
+        $client = $this->client();
+
+        $id = Id\generateUniqueId();
+        self::respond($client, "{$id}.scatter", ['a', 'b']);
+
+        $replies = $client->requestMany("{$id}.scatter", new Message('?'), timeout: 5);
+        foreach ($replies as $_) {
+            $replies->stop();
+        }
+
+        // Only the responders' subscriptions are left.
+        self::assertCount(2, self::subscribers($client));
+    }
+
+    public function testRequestManyRequiresEndCondition(): void
+    {
+        $client = $this->client();
+
+        $this->expectException(\InvalidArgumentException::class);
+        $client->requestMany(Id\generateUniqueId() . '.scatter', new Message('?'));
+    }
+
+    /**
+     * @param array{stall?: float, timeout?: float} $arguments
+     */
+    #[TestWith([['timeout' => -1.0]])]
+    #[TestWith([['timeout' => 0.0]])]
+    #[TestWith([['timeout' => INF]])]
+    #[TestWith([['timeout' => NAN]])]
+    #[TestWith([['timeout' => 1.0, 'stall' => 0.0]])]
+    #[TestWith([['timeout' => 1.0, 'stall' => INF]])]
+    public function testRequestManyRejectsInvalidDurationsBeforeSubscribing(array $arguments): void
+    {
+        $client = $this->client();
+
+        try {
+            $client->requestMany(Id\generateUniqueId() . '.scatter', new Message('?'), ...$arguments);
+            self::fail('The durations are invalid, the request must be refused.');
+        } catch (\InvalidArgumentException) {
+        }
+
+        self::assertSame([], self::subscribers($client));
+    }
+
+    #[TestWith([0])]
+    #[TestWith([-1])]
+    public function testRequestManyRejectsNonPositiveMaxMessagesBeforeSubscribing(int $maxMessages): void
+    {
+        $client = $this->client();
+
+        try {
+            /** @phpstan-ignore argument.type */
+            $client->requestMany(Id\generateUniqueId() . '.scatter', new Message('?'), maxMessages: $maxMessages, timeout: 1);
+            self::fail('The maxMessages is invalid, the request must be refused.');
+        } catch (\InvalidArgumentException) {
+        }
+
+        self::assertSame([], self::subscribers($client));
+    }
+
+    #[TestWith([-1.0])]
+    #[TestWith([0.0])]
+    #[TestWith([INF])]
+    #[TestWith([NAN])]
+    public function testRequestRejectsInvalidTimeout(float $timeout): void
+    {
+        $client = $this->client();
+
+        try {
+            $client->request(Id\generateUniqueId() . '.nobody', new Message('?'), timeout: $timeout);
+            self::fail('The timeout is invalid, the request must be refused.');
+        } catch (\InvalidArgumentException) {
+        }
+
+        self::assertSame(0, self::pendingRequests($client));
+        self::assertSame([], self::subscribers($client));
+    }
+
+    public function testStallAloneIsNotEndCondition(): void
+    {
+        $client = $this->client();
+
+        $this->expectException(\InvalidArgumentException::class);
+        $client->requestMany(Id\generateUniqueId() . '.scatter', new Message('?'), stall: 0.05);
+    }
+
+    public function testSilentResponderIsEndedByTimeoutNotStall(): void
+    {
+        $client = $this->client();
+
+        $id = Id\generateUniqueId();
+        $client->subscribe("{$id}.silent", static fn() => null);
+
+        $started = microtime(true);
+        $replies = self::payloads($client->requestMany("{$id}.silent", new Message('?'), stall: 0.05, timeout: 0.3));
+
+        self::assertSame([], $replies);
+        self::assertGreaterThanOrEqual(0.29, microtime(true) - $started, 'The stall does not run before the first reply.');
+    }
+
+    public function testRequestManyEndsByConfigTimeout(): void
+    {
+        $config = $this->config();
+        $client = $this->clientWithConfig(new Config(
+            urls: $config->urls,
+            user: $config->user,
+            password: $config->password,
+            noResponders: $config->noResponders,
+            requestTimeout: 0.2,
+        ));
+
+        $id = Id\generateUniqueId();
+        self::respond($client, "{$id}.scatter", ['a', 'b']);
+
+        self::assertEqualsCanonicalizing(['a', 'b'], self::payloads($client->requestMany("{$id}.scatter", new Message('?'))));
+    }
+
+    /**
+     * @param non-empty-string $subject
+     * @param list<non-empty-string> $answers one responder per answer
+     */
+    private static function respond(Client $client, string $subject, array $answers): void
+    {
+        foreach ($answers as $answer) {
+            $client->subscribe($subject, static function (Delivery $delivery) use ($answer): void {
+                $delivery->reply(new Message($answer));
+            });
+        }
+    }
+
+    /**
+     * @param iterable<Delivery> $deliveries
+     * @return list<?string>
+     */
+    private static function payloads(iterable $deliveries): array
+    {
+        return async(static function () use ($deliveries): array {
+            $payloads = [];
+            foreach ($deliveries as $delivery) {
+                $payloads[] = $delivery->message->payload;
+            }
+
+            return $payloads;
+        })->await(new TimeoutCancellation(5));
+    }
+
+    /**
+     * @return array<array-key, mixed>
+     */
+    private static function subscribers(Client $client): array
+    {
+        /** @var array<array-key, mixed> */
+        return (new \ReflectionProperty(Client::class, 'subscribers'))->getValue($client);
+    }
+}
diff --git a/tests/SubscriptionTest.php b/tests/SubscriptionTest.php
new file mode 100644
index 0000000..8cf859c
--- /dev/null
+++ b/tests/SubscriptionTest.php
@@ -0,0 +1,288 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Thesis\Nats;
+
+use Amp\DeferredFuture;
+use Amp\TimeoutCancellation;
+use PHPUnit\Framework\Attributes\CoversClass;
+use PHPUnit\Framework\Attributes\TestWith;
+use Thesis\Nats\Exception\SlowConsumer;
+use Thesis\Nats\Internal\Id;
+use function Amp\delay;
+
+#[CoversClass(Subscription::class)]
+#[CoversClass(Client::class)]
+final class SubscriptionTest extends NatsTestCase
+{
+    public function testSubscriptionExposesSubjectAndQueueGroup(): void
+    {
+        $client = $this->client();
+
+        $id = Id\generateUniqueId();
+
+        $plain = $client->subscribe("{$id}.*", static fn() => null);
+        self::assertSame("{$id}.*", $plain->subject);
+        self::assertNull($plain->queueGroup);
+
+        $queued = $client->subscribe("{$id}.>", static fn() => null, queueGroup: 'workers');
+        self::assertSame("{$id}.>", $queued->subject);
+        self::assertSame('workers', $queued->queueGroup);
+    }
+
+    public function testSubscriptionCompletesAfterMaxMessages(): void
+    {
+        $client = $this->client();
+
+        $id = Id\generateUniqueId();
+
+        /** @var \ArrayObject<int, ?string> $received */
+        $received = new \ArrayObject();
+
+        $subscription = $client->subscribe(
+            $id,
+            static function (Delivery $delivery) use ($received): void {
+                $received[] = $delivery->message->payload;
+            },
+            max: 2,
+        );
+
+        foreach (['1', '2', '3', '4'] as $payload) {
+            $client->publish($id, new Message($payload));
+        }
+
+        $subscription->awaitCompletion(new TimeoutCancellation(5));
+        self::assertSame(['1', '2'], $received->getArrayCopy());
+        self::assertSame(2, $subscription->stats()->delivered);
+
+        // The server has removed the subscription, so nothing is received anymore.
+        $client->publish($id, new Message('5'));
+        delay(0.1);
+
+        self::assertSame(['1', '2'], $received->getArrayCopy());
+        self::assertSame([], self::subscribers($client));
+    }
+
+    public function testIteratorEndsAfterMaxMessages(): void
+    {
+        $client = $this->client();
+
+        $id = Id\generateUniqueId();
+        $deliveries = $client->subscribeIterator($id, max: 3);
+
+        foreach (['1', '2', '3', '4', '5'] as $payload) {
+            $client->publish($id, new Message($payload));
+        }
+
+        $received = [];
+        foreach ($deliveries as $delivery) {
+            $received[] = $delivery->message->payload;
+        }
+
+        self::assertSame(['1', '2', '3'], $received);
+        self::assertSame([], self::subscribers($client));
+    }
+
+    #[TestWith([0])]
+    #[TestWith([-1])]
+    public function testSubscribeRejectsNonPositiveMax(int $max): void
+    {
+        $client = $this->client();
+
+        try {
+            /** @phpstan-ignore argument.type */
+            $client->subscribe(Id\generateUniqueId(), static fn() => null, max: $max);
+            self::fail('The max is invalid, the subscription must be refused.');
+        } catch (\InvalidArgumentException) {
+        }
+
+        self::assertSame([], self::subscribers($client));
+    }
+
+    public function testStopBeforeMaxMessagesUnsubscribes(): void
+    {
+        $client = $this->client();
+
+        $id = Id\generateUniqueId();
+        $subscription = $client->subscribe($id, static fn() => null, max: 10);
+
+        $client->publish($id, new Message('1'));
+        $subscription->stop();
+        $subscription->awaitCompletion(new TimeoutCancellation(5));
+
+        self::assertSame([], self::subscribers($client));
+    }
+
+    public function testSlowConsumerDropsMessagesBeyondPendingLimits(): void
+    {
+        $client = $this->client();
+
+        /** @var \ArrayObject<int, \Throwable> $errors */
+        $errors = new \ArrayObject();
+        $client->onError(static function (\Throwable $e) use ($errors): void {
+            $errors[] = $e;
+        });
+
+        $id = Id\generateUniqueId();
+
+        /** @var DeferredFuture<void> $release */
+        $release = new DeferredFuture();
+
+        /** @var \ArrayObject<int, ?string> $received */
+        $received = new \ArrayObject();
+
+        $subscription = $client->subscribe(
+            $id,
+            static function (Delivery $delivery) use (
+                $release,
+                $received,
+            ): void {
+                $received[] = $delivery->message->payload;
+                $release->getFuture()->await();
+            },
+            pendingLimits: new PendingLimits(messages: 2, bytes: null),
+        );
+
+        // The handler is busy with the first message, the rest pile up.
+        $client->publish($id, new Message('1'));
+        self::waitFor(static fn() => $subscription->stats()->delivered === 1);
+
+        foreach (['2', '3', '4', '5'] as $payload) {
+            $client->publish($id, new Message($payload));
+        }
+
+        self::waitFor(static fn() => $subscription->stats()->dropped === 2);
+
+        $stats = $subscription->stats();
+        self::assertSame(1, $stats->delivered);
+        self::assertSame(2, $stats->pendingMessages);
+        self::assertSame(2, $stats->pendingBytes);
+
+        self::assertCount(1, $errors, 'A slow consumer is reported once per episode.');
+        self::assertInstanceOf(SlowConsumer::class, $errors[0]);
+        self::assertSame($subscription, $errors[0]->subscription);
+
+        $release->complete();
+        self::waitFor(static fn() => $subscription->stats()->pendingMessages === 0);
+
+        self::assertSame(['1', '2', '3'], $received->getArrayCopy());
+        self::assertEquals(new SubscriptionStats(pendingMessages: 0, pendingBytes: 0, delivered: 3, dropped: 2), $subscription->stats());
+
+        // Caught up: the next overflow is a new episode.
+        $client->publish($id, new Message('6'));
+        self::waitFor(static fn() => $subscription->stats()->delivered === 4);
+        self::assertCount(1, $errors);
+    }
+
+    public function testPendingBytesLimit(): void
+    {
+        $client = $this->client();
+        $client->onError(static fn() => null);
+
+        $id = Id\generateUniqueId();
+
+        /** @var DeferredFuture<void> $release */
+        $release = new DeferredFuture();
+
+        $subscription = $client->subscribe(
+            $id,
+            static fn() => $release->getFuture()->await(),
+            pendingLimits: new PendingLimits(messages: null, bytes: 10),
+        );
+
+        $client->publish($id, new Message('first'));
+        self::waitFor(static fn() => $subscription->stats()->delivered === 1);
+
+        foreach (['12345', '67890', 'x'] as $payload) {
+            $client->publish($id, new Message($payload));
+        }
+
+        self::waitFor(static fn() => $subscription->stats()->dropped === 1);
+
+        // 'first' is in the handler, '12345' and '67890' fill the 10 bytes, 'x' does not fit.
+        self::assertSame(10, $subscription->stats()->pendingBytes);
+
+        $release->complete();
+    }
+
+    public function testIteratorDropsMessagesBeyondPendingLimits(): void
+    {
+        $client = $this->client();
+
+        /** @var DeferredFuture<\Throwable> $error */
+        $error = new DeferredFuture();
+        $client->onError(static function (\Throwable $e) use ($error): void {
+            if (!$error->isComplete()) {
+                $error->complete($e);
+            }
+        });
+
+        $id = Id\generateUniqueId();
+
+        // Nobody iterates: the buffer fills up, then the pending messages reach the limit.
+        $deliveries = $client->subscribeIterator($id, bufferSize: 1, pendingLimits: new PendingLimits(messages: 1, bytes: null));
+
+        foreach (['1', '2', '3', '4', '5'] as $payload) {
+            $client->publish($id, new Message($payload));
+        }
+
+        $e = $error->getFuture()->await(new TimeoutCancellation(5));
+        self::assertInstanceOf(SlowConsumer::class, $e);
+        self::assertSame($id, $e->subscription?->subject);
+
+        $deliveries->stop();
+    }
+
+    public function testStatsAreResetOnStop(): void
+    {
+        $client = $this->client();
+
+        $id = Id\generateUniqueId();
+
+        /** @var DeferredFuture<void> $release */
+        $release = new DeferredFuture();
+
+        $subscription = $client->subscribe($id, static fn() => $release->getFuture()->await());
+
+        $client->publish($id, new Message('1'));
+        self::waitFor(static fn() => $subscription->stats()->delivered === 1);
+
+        foreach (['2', '3'] as $payload) {
+            $client->publish($id, new Message($payload));
+        }
+
+        self::waitFor(static fn() => $subscription->stats()->pendingMessages === 2);
+
+        $release->complete();
+        $subscription->stop();
+        self::waitFor(static fn() => $subscription->stats()->pendingMessages === 0);
+
+        self::assertSame(0, $subscription->stats()->pendingBytes);
+    }
+
+    /**
+     * @param \Closure(): bool $condition
+     */
+    private static function waitFor(\Closure $condition, float $timeout = 5): void
+    {
+        $deadline = microtime(true) + $timeout;
+
+        while (!$condition()) {
+            if (microtime(true) > $deadline) {
+                self::fail('The condition was not met in time.');
+            }
+
+            delay(0.01);
+        }
+    }
+
+    /**
+     * @return array<array-key, mixed>
+     */
+    private static function subscribers(Client $client): array
+    {
+        /** @var array<array-key, mixed> */
+        return (new \ReflectionProperty(Client::class, 'subscribers'))->getValue($client);
+    }
+}
diff --git a/tests/ValidationTest.php b/tests/ValidationTest.php
index 6617eea..7b7a567 100644
--- a/tests/ValidationTest.php
+++ b/tests/ValidationTest.php
@@ -45,12 +45,16 @@ public function testInvalidCallFailsBeforeConnecting(\Closure $call, string $exc
         $server = listen('127.0.0.1:0');
         $connections = 0;
 
-        async(static function () use ($server, &$connections): void {
+        $future = async(static function () use (
+            $server,
+            &$connections,
+        ): void {
             while (($socket = $server->accept()) !== null) {
                 ++$connections;
                 $socket->close();
             }
-        })->ignore();
+        });
+        $future->ignore();
 
         try {
             $nc = $this->clientWithUrls([self::address($server)]);
diff --git a/tests/VerboseModeTest.php b/tests/VerboseModeTest.php
index 15839c1..366fa85 100644
--- a/tests/VerboseModeTest.php
+++ b/tests/VerboseModeTest.php
@@ -80,7 +80,10 @@ public function testCancelledPublishKeepsAcknowledgementOrder(): void
         $received = [];
         /** @var DeferredFuture<void> $done */
         $done = new DeferredFuture();
-        $client->subscribe("{$id}.*", static function (Delivery $delivery) use (&$received, $done): void {
+        $client->subscribe("{$id}.*", static function (Delivery $delivery) use (
+            &$received,
+            $done,
+        ): void {
             $received[] = $delivery->message->payload ?? '';
             if (\count($received) === 2) {
                 $done->complete();
@@ -195,6 +198,50 @@ public function testPingsAreNotAwaitedAsAcknowledgedCommands(): void
         $client->stop();
     }
 
+    public function testSubscribeWithMaxAwaitsBothAcknowledgements(): void
+    {
+        $client = $this->verboseClient(user: 'restricted', password: 'Pswd1');
+        $id = Id\generateUniqueId();
+
+        /** @var \ArrayObject<int, ?string> $received */
+        $received = new \ArrayObject();
+
+        $subscription = $client->subscribe(
+            "allowed.{$id}",
+            static function (Delivery $delivery) use ($received): void {
+                $received[] = $delivery->message->payload;
+            },
+            cancellation: new TimeoutCancellation(5),
+            max: 2,
+        );
+
+        foreach (['1', '2', '3'] as $payload) {
+            $client->publish("allowed.{$id}", new Message($payload), cancellation: new TimeoutCancellation(5));
+        }
+
+        $subscription->awaitCompletion(new TimeoutCancellation(5));
+        self::assertSame(['1', '2'], $received->getArrayCopy());
+
+        // SUB is rejected, the +OK of the UNSUB written along with it must not go to the next command.
+        try {
+            $client->subscribe("forbidden.{$id}", static fn() => null, cancellation: new TimeoutCancellation(5), max: 1);
+            self::fail('The subscription is not permitted, it must fail.');
+        } catch (PermissionViolated $e) {
+            self::assertStringContainsString("forbidden.{$id}", $e->getMessage());
+        }
+
+        try {
+            $client->publish("forbidden.{$id}", new Message('ok'), cancellation: new TimeoutCancellation(5));
+            self::fail('The -ERR must be attributed to the publish into the forbidden subject.');
+        } catch (PermissionViolated $e) {
+            self::assertStringContainsString("forbidden.{$id}", $e->getMessage());
+        }
+
+        $client->publish("allowed.{$id}", new Message('ok'), cancellation: new TimeoutCancellation(5));
+
+        $client->stop();
+    }
+
     private function proxy(): PausableProxy
     {
         return $this->proxy ??= PausableProxy::start($this->natsAddress());