From 827db76537e4d6459681cb1bc96c5deb96243134 Mon Sep 17 00:00:00 2001 From: Patrick Hemmer Date: Wed, 13 May 2026 23:36:51 -0400 Subject: [PATCH 1/2] nats-py: Fix PullSubscription.fetch hang due to orphan lingering request This fixes an issue in the nats-py code where a race condition occurs that leaves an orphaned lingering request on the server, causing `fetch` to hang until the timeout expires. fixes #933 --- nats/src/nats/js/client.py | 29 ++++++++++++++++++- nats/tests/test_js.py | 59 ++++++++++++++++++++++++++++++++++++++ 2 files changed, 87 insertions(+), 1 deletion(-) diff --git a/nats/src/nats/js/client.py b/nats/src/nats/js/client.py index 66279635f..a6b0fc83a 100644 --- a/nats/src/nats/js/client.py +++ b/nats/src/nats/js/client.py @@ -1265,9 +1265,29 @@ async def _fetch_n( # Second request: lingering request that will block until new messages # are made available and delivered to the client. + # + # Use the *remaining* deadline as the request's expires rather than + # the original full timeout. The original expires was computed at + # the very start of fetch() and may be nearly exhausted by the time + # we reach this point (e.g. when the server's 408 for the no-wait + # probe arrives just before the asyncio timer fires). Sending a + # lingering request with the full original expires in that situation + # creates an orphaned pull request that survives on the server long + # after the client has timed out, capturing the next published + # message and causing the subsequent fetch() to stall for the full + # timeout window. + deadline = JetStreamContext._time_until(timeout, start_time) + if deadline is not None and deadline <= 0: + raise asyncio.TimeoutError + next_req = {} next_req["batch"] = needed - if expires: + if deadline is not None: + remaining_expires = int(deadline * 1_000_000_000) - 100_000 + if remaining_expires <= 0: + raise asyncio.TimeoutError + next_req["expires"] = remaining_expires + elif expires: next_req["expires"] = expires if heartbeat: next_req["idle_heartbeat"] = int(heartbeat * 1_000_000_000) # to nanoseconds @@ -1335,6 +1355,13 @@ async def _fetch_n( if JetStreamContext._is_heartbeat(status): got_any_response = True continue + if status in ( + api.StatusCode.NO_MESSAGES, + api.StatusCode.REQUEST_TIMEOUT, + ): + # No more messages will be delivered on this pull + # request; return what we have. + break if JetStreamContext._is_processable_msg(status, msg): needed -= 1 msgs.append(msg) diff --git a/nats/tests/test_js.py b/nats/tests/test_js.py index 83b00d7ae..110ea0e4b 100644 --- a/nats/tests/test_js.py +++ b/nats/tests/test_js.py @@ -1173,6 +1173,65 @@ async def test_fetch_heartbeats(self): await nc.close() + @async_long_test + async def test_fetch_no_orphan_on_timeout(self): + """ + fetch() must not leave an orphaned pull request on the server when it + times out. + + When the server's 408 REQUEST_TIMEOUT (sent at expires = timeout - + 100µs) arrives before Python's asyncio timer fires, _fetch_n sends a + second "lingering" pull request with the full original expires and then + immediately abandons it as the asyncio timer fires. That lingering + remains on the server as an orphan. + + On the next fetch() call the server has two outstanding pull requests. + NATS routes incoming messages to the oldest one — the orphan. The + current fetch()'s probe sees no delivery and must wait for its own + expires to elapse (~timeout seconds) before returning the one message + it already holds in hand. + """ + nc = NATS() + await nc.connect() + + js = nc.jetstream() + await js.add_stream(name="TEST_ORPHAN", subjects=["orphan.>"]) + sub = await js.pull_subscribe("orphan.>", "durable-orphan") + + # First fetch on an empty stream with a short timeout. The server's + # 408 (sent at expires = 100ms - 100µs) arrives before Python's asyncio + # timer, causing _fetch_n to send an orphaned lingering pull request + # that remains on the server after the client times out. + try: + await sub.fetch(100, timeout=0.1) + except (nats.errors.TimeoutError, asyncio.TimeoutError): + pass + + # Start a new fetch, then publish one message after a brief pause. + # Without the fix the orphan captures the message and the current + # fetch's probe must wait out its full timeout (~3 s) before returning. + # With the fix no orphan exists and the message is returned promptly. + fetch_task = asyncio.create_task(sub.fetch(100, timeout=3.0)) + await asyncio.sleep(0.05) + + await js.publish("orphan.test", b"hello") + t0 = time.monotonic() + msgs = await fetch_task + elapsed = time.monotonic() - t0 + + assert len(msgs) == 1 + assert msgs[0].data == b"hello" + for msg in msgs: + await msg.ack() + + assert elapsed < 1.0, ( + f"fetch() returned {elapsed:.3f}s after publish — expected < 1s. " + "An orphaned pull request likely captured the message, forcing the " + "current fetch to stall until the probe's own expires elapsed." + ) + + await nc.close() + @async_long_test async def test_subscribe_filter_subjects(self): nc = NATS() From ab04fa6d4371ead8b0a4d30fb9745336b0d41cb7 Mon Sep 17 00:00:00 2001 From: Patrick Hemmer Date: Thu, 14 May 2026 13:55:14 -0400 Subject: [PATCH 2/2] fix PullSubscription.fetch hang when messages are buffered MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit When _fetch_n drains messages from _pending_queue at the start of a fetch, it then sends a no_wait probe to collect more from the server. The probe includes an expires field, which causes NATS server ≥ 2.10 to ignore no_wait and treat the request as a regular lingering pull. If the stream has no further messages available (they were already delivered into the queue before fetch() ran), the probe blocks for the full expires duration before returning — stalling the caller even though the drained messages are ready to return immediately. This fixes the issue by omitting expires from the no_wait probe when the drain step already collected messages. Without expires, the server correctly honors no_wait and responds immediately with any available messages or a 404. When the drain step found nothing, expires is still included so that the existing server-side lingering behaviour is preserved. --- nats/src/nats/js/client.py | 10 +++- nats/tests/test_js.py | 99 ++++++++++++++++++++++++++++++++++++++ 2 files changed, 108 insertions(+), 1 deletion(-) diff --git a/nats/src/nats/js/client.py b/nats/src/nats/js/client.py index a6b0fc83a..33f1c60cd 100644 --- a/nats/src/nats/js/client.py +++ b/nats/src/nats/js/client.py @@ -1203,9 +1203,17 @@ async def _fetch_n( # First request: Use no_wait to synchronously get as many available # based on the batch size until server sends 'No Messages' status msg. + # Omit `expires` when the drain step already found messages: NATS + # server ignores no_wait when expires is present, treating the probe + # as a lingering pull and blocking for the full expires duration. + # Without expires the server honors no_wait immediately, so Phase 3 + # returns quickly with any additional server-side messages or a 404, + # and the existing `if len(msgs) > 0` guard returns the collected + # messages without delay. When the drain step found nothing expires + # is still included to preserve the intended behaviour. next_req = {} next_req["batch"] = needed - if expires: + if expires and not msgs: next_req["expires"] = expires if heartbeat: next_req["idle_heartbeat"] = int(heartbeat * 1_000_000_000) # to nanoseconds diff --git a/nats/tests/test_js.py b/nats/tests/test_js.py index 110ea0e4b..360af5ff7 100644 --- a/nats/tests/test_js.py +++ b/nats/tests/test_js.py @@ -1232,6 +1232,105 @@ async def test_fetch_no_orphan_on_timeout(self): await nc.close() + @async_long_test + async def test_fetch_returns_promptly_with_pending_queue_messages(self): + """ + fetch() must return promptly when messages are already buffered in + the subscription's internal pending queue before fetch() is called. + + _fetch_n drains _pending_queue first, then sends a no_wait probe for + the remaining batch slots. If the probe includes an `expires` field, + NATS server 2.12.6 ignores no_wait and treats the request as a + lingering pull, blocking for the full expires duration even though the + message was already collected during the drain step. + """ + nc = NATS() + await nc.connect() + + js = nc.jetstream() + await js.add_stream(name="TEST_DRAIN", subjects=["drain.>"]) + sub = await js.pull_subscribe("drain.>", "durable-drain") + + # Publish a message, then deliver it into the subscription's + # _pending_queue via a direct no_wait probe — bypassing _fetch_n so + # the message is already buffered before fetch() is called. + await js.publish("drain.test", b"hello") + await sub._nc.publish( + sub._nms, + json.dumps({"batch": 1, "no_wait": True}).encode(), + sub._deliver, + ) + await asyncio.sleep(0.1) + + assert not sub._sub._pending_queue.empty(), "message did not arrive in _pending_queue — test setup failed" + + # fetch() should drain the queued message and return without waiting + # for the no_wait probe's expires to elapse (~5 s). + t0 = time.monotonic() + msgs = await sub.fetch(100, timeout=5.0) + elapsed = time.monotonic() - t0 + + assert len(msgs) == 1 + assert msgs[0].data == b"hello" + for msg in msgs: + await msg.ack() + + assert elapsed < 1.0, ( + f"fetch() took {elapsed:.3f}s to return a message that was already " + "in _pending_queue; expected < 1s. The no_wait probe likely included " + "an `expires` field that caused the server to treat it as a lingering " + "pull, blocking until the probe timed out." + ) + + await nc.close() + + @async_long_test + async def test_fetch_collects_server_messages_alongside_pending_queue(self): + """ + fetch() must collect messages from both _pending_queue and the server + in a single call. + + If the drain step picks up messages from _pending_queue and then + returns immediately without sending the no_wait probe, any messages + sitting in the stream on the server side are silently skipped until + the next fetch() call. + """ + nc = NATS() + await nc.connect() + + js = nc.jetstream() + await js.add_stream(name="TEST_DRAIN2", subjects=["drain2.>"]) + sub = await js.pull_subscribe("drain2.>", "durable-drain2") + + # Publish two messages. + await js.publish("drain2.test", b"msg-a") + await js.publish("drain2.test", b"msg-b") + + # Deliver only the first message into _pending_queue via a direct + # no_wait probe, bypassing _fetch_n. msg-b remains on the server. + await sub._nc.publish( + sub._nms, + json.dumps({"batch": 1, "no_wait": True}).encode(), + sub._deliver, + ) + await asyncio.sleep(0.1) + + assert not sub._sub._pending_queue.empty(), "msg-a did not arrive in _pending_queue — test setup failed" + + # fetch() should drain msg-a from the queue AND collect msg-b from + # the server via the no_wait probe, returning both in one call. + msgs = await sub.fetch(100, timeout=2.0) + + assert len(msgs) == 2, ( + f"expected 2 messages (one from _pending_queue, one from server) " + f"but got {len(msgs)}. fetch() likely returned after the drain step " + "without sending the no_wait probe to the server." + ) + for msg in msgs: + await msg.ack() + + await nc.close() + @async_long_test async def test_subscribe_filter_subjects(self): nc = NATS()