Stop holding a fiber per pending message behind a slow subscription handler

README.md

@@ -288,8 +288,12 @@ The messages a subscription has received, but its handler has not taken yet, are
 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.
+The client never stops reading the connection for a slow handler, so that it keeps answering the server's pings, the replies to the requests
+the handler makes and the messages of the other subscriptions. Hence the pending limits are what bounds the memory the pending messages take,
+and `PendingLimits::unlimited()` lets them grow with no bound.
+
+The `bufferSize` of `subscribeIterator()` and `requestMany()` is a different thing: it is how many messages the iterator holds ahead of the consumer,
+they are not pending anymore. A message that does not fit stays pending until the consumer takes one, so the pending limits are reached only behind a full buffer.
 
 ```php
 <?php

src/Client.php

@@ -156,7 +156,7 @@ public function rawPublish(
     /**
      * @param non-empty-string $subject
      * @param ?non-empty-string $queueGroup
-     * @param positive-int $bufferSize
+     * @param positive-int $bufferSize how many deliveries the iterator holds ahead of the consumer, see {@see PendingLimits}
      * @param ?positive-int $max the number of messages after which the subscription ends, see {@see subscribe()}
      * @return Iterator<Delivery>
      * @throws NatsException
@@ -189,18 +189,16 @@ public function subscribeIterator(
      * @param non-empty-string $subject
      * @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
+     * @param PendingLimits $pendingLimits the messages the handler has not taken yet, the ones beyond them are dropped
      * @throws NatsException
      * @throws \Throwable
      */
     public function subscribe(
         string $subject,
         callable $handler,
         ?string $queueGroup = null,
-        int $bufferSize = 1_000,
         ?Cancellation $cancellation = null,
         ?int $max = null,
         PendingLimits $pendingLimits = new PendingLimits(),
@@ -215,7 +213,7 @@ public function subscribe(
             Validate::count($max, 'max');
         }
 
-        return $this->register($subject, $handler, $queueGroup, $bufferSize, $cancellation, $max, $pendingLimits);
+        return $this->register($subject, $handler, $queueGroup, $cancellation, $max, $pendingLimits);
     }
 
     /**
@@ -291,7 +289,6 @@ public function requestMany(
         $subscription = $this->register(
             subject: $inbox = Id\generateInboxId(),
             handler: $handler,
-            bufferSize: $bufferSize,
             cancellation: $cancellation,
             max: $maxMessages,
             onReceive: $handler->received(...),
@@ -440,7 +437,6 @@ public function __destruct()
      * @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
@@ -450,7 +446,6 @@ 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(),
@@ -469,7 +464,6 @@ private function register(
                 $unsubscribe($subscriptionId);
             },
             handler: $handler,
-            bufferSize: $bufferSize,
             onSlowConsumer: $this->reportError(...),
             remaining: $max,
             onReceive: $onReceive,

src/Internal/Subscription/Backlog.php

@@ -12,8 +12,8 @@
 /**
  * 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.
+ * from the buffer size of an iterator. The limits are the only thing that bounds the memory the pending messages take:
+ * the connection never waits for a slow handler, so that it keeps answering the server and serving the other subscriptions.
  *
  * @internal
  */

src/Internal/Subscription/SubscriptionHandler.php

@@ -33,7 +33,6 @@ final class SubscriptionHandler
      * @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
@@ -43,16 +42,18 @@ public function __construct(
         ?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(...));
 
+        // Unbounded, so that a push never suspends the fiber the connection dispatches the message in: a slow handler
+        // must neither hold a fiber per pending message nor stall the connection. The backlog limits bound the queue.
+        // Half of the int range, since amp adds the buffer size to the queue position.
         /** @var Pipeline\Queue<Delivery> $queue */
-        $queue = new Pipeline\Queue($bufferSize);
+        $queue = new Pipeline\Queue(PHP_INT_MAX >> 1);
 
         /** @var DeferredFuture<Operation<*>> */
         $completeSubscriptionDeferred = new DeferredFuture();
@@ -194,7 +195,6 @@ private function accept(Delivery $delivery): bool
             return true;
         }
 
-        // 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;
         }

src/JetStream/PullConsumer.php

@@ -67,7 +67,6 @@ public function fetch(FetchConfig $config, ?Cancellation $cancellation = null):
         $subscription = $this->nats->subscribe(
             subject: $reply = Id\generateInboxId(),
             handler: $handler,
-            bufferSize: $config->batch,
             cancellation: $cancellation,
         );
 

src/PendingLimits.php

@@ -11,9 +11,13 @@
  * 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.
+ * The limits are what bounds the memory a slow handler makes the pending messages take: the client never stops reading
+ * the connection for it, so that the server keeps getting answered and the other subscriptions keep getting messages,
+ * as in nats.go. With {@see unlimited()} nothing bounds that memory.
+ *
+ * The buffer size of {@see Client::subscribeIterator()} and {@see Client::requestMany()} is a different thing:
+ * it is how many messages the iterator holds ahead of the consumer, they are not pending anymore. A message that does
+ * not fit stays pending until the consumer takes one, so the limits are reached only behind a full buffer.
  *
  * @api
  */

tests/SubscriptionTest.php

@@ -234,6 +234,38 @@ public function testIteratorDropsMessagesBeyondPendingLimits(): void
         $deliveries->stop();
     }
 
+    public function testPendingMessagesBehindSlowHandlerDoNotHoldFibers(): 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('first'));
+        self::waitFor(static fn() => $subscription->stats()->delivered === 1);
+
+        $messages = 5_000;
+        $before = memory_get_usage();
+
+        for ($i = 0; $i < $messages; ++$i) {
+            $client->publish($id, new Message('x'));
+        }
+
+        self::waitFor(static fn() => $subscription->stats()->pendingMessages === $messages, timeout: 10);
+
+        // A suspended fiber alone takes about 16 KiB, a pending message without one takes a fraction of that.
+        self::assertLessThan(4 * 1024, intdiv(memory_get_usage() - $before, $messages));
+
+        $release->complete();
+        self::waitFor(static fn() => $subscription->stats()->delivered === $messages + 1, timeout: 10);
+
+        $subscription->stop();
+    }
+
     public function testStatsAreResetOnStop(): void
     {
         $client = $this->client();