From e7af20b5985106faa4cbf8d31a21cafba702f607 Mon Sep 17 00:00:00 2001 From: kafkiansky Date: Thu, 15 Oct 2026 20:50:30 +0300 Subject: [PATCH] Stop holding a fiber per pending message behind a slow subscription handler --- README.md | 8 +++-- src/Client.php | 12 ++----- src/Internal/Subscription/Backlog.php | 4 +-- .../Subscription/SubscriptionHandler.php | 8 ++--- src/JetStream/PullConsumer.php | 1 - src/PendingLimits.php | 10 ++++-- tests/SubscriptionTest.php | 32 +++++++++++++++++++ 7 files changed, 54 insertions(+), 21 deletions(-) diff --git a/README.md b/README.md index c9ab1d0..3b2d034 100644 --- a/README.md +++ b/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 * @throws NatsException @@ -189,10 +189,9 @@ 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 ), * 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 */ @@ -200,7 +199,6 @@ 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, diff --git a/src/Internal/Subscription/Backlog.php b/src/Internal/Subscription/Backlog.php index e836304..72a6230 100644 --- a/src/Internal/Subscription/Backlog.php +++ b/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 */ diff --git a/src/Internal/Subscription/SubscriptionHandler.php b/src/Internal/Subscription/SubscriptionHandler.php index d723e72..368ebbc 100644 --- a/src/Internal/Subscription/SubscriptionHandler.php +++ b/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,7 +42,6 @@ public function __construct( ?string $queueGroup, \Closure $unsubscribe, callable $handler, - int $bufferSize, private readonly \Closure $onSlowConsumer, private ?int $remaining = null, private readonly ?\Closure $onReceive = null, @@ -51,8 +49,11 @@ public function __construct( ) { $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 $queue */ - $queue = new Pipeline\Queue($bufferSize); + $queue = new Pipeline\Queue(PHP_INT_MAX >> 1); /** @var DeferredFuture> */ $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; } diff --git a/src/JetStream/PullConsumer.php b/src/JetStream/PullConsumer.php index 1331542..ff2d9c6 100644 --- a/src/JetStream/PullConsumer.php +++ b/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, ); diff --git a/src/PendingLimits.php b/src/PendingLimits.php index 0cbe007..40ca400 100644 --- a/src/PendingLimits.php +++ b/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 */ diff --git a/tests/SubscriptionTest.php b/tests/SubscriptionTest.php index 7c94652..7536382 100644 --- a/tests/SubscriptionTest.php +++ b/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 $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();