Fix PullSubscription.fetch hang due to orphan lingering request

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)

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()