From 77c381b456970973f584f36ea503516ad228c48b Mon Sep 17 00:00:00 2001 From: Casper Beyer Date: Tue, 14 Jul 2026 05:05:20 +0200 Subject: [PATCH 1/2] Raise publish_async ack errors on the returned future APIError.from_error raises instead of returning, so the error branch in _handle_async_reply threw before set_exception ran. The exception escaped into the delivery task (logged as "nats: encountered error") and the caller's ack future was left pending forever. Capture the raised error and attach it to the future so awaiting the ack surfaces it. --- nats/src/nats/js/client.py | 9 +++++++-- nats/tests/test_js.py | 17 +++++++++++++++++ 2 files changed, 24 insertions(+), 2 deletions(-) diff --git a/nats/src/nats/js/client.py b/nats/src/nats/js/client.py index d516a6e2..43936e56 100644 --- a/nats/src/nats/js/client.py +++ b/nats/src/nats/js/client.py @@ -165,8 +165,13 @@ async def _handle_async_reply(self, msg: Msg) -> None: try: resp = json.loads(msg.data) if "error" in resp: - err = nats.js.errors.APIError.from_error(resp["error"]) - future.set_exception(err) + # APIError.from_error raises rather than returning, so capture + # the constructed error and attach it to the future instead of + # letting it escape and leave the future pending forever. + try: + raise nats.js.errors.APIError.from_error(resp["error"]) + except nats.js.errors.APIError as err: + future.set_exception(err) return ack = api.PubAck.from_response(resp) diff --git a/nats/tests/test_js.py b/nats/tests/test_js.py index fd5bc1ce..31325499 100644 --- a/nats/tests/test_js.py +++ b/nats/tests/test_js.py @@ -193,6 +193,23 @@ async def test_publish_async(self): await nc.close() + @async_test + async def test_publish_async_error_ack_is_raised(self): + # Regression for #985: a server error ack must be raised on the + # returned future rather than being swallowed (logged and left + # pending forever). + nc = NATS() + await nc.connect() + js = nc.jetstream() + await js.add_stream(name="AERR", subjects=["aerr"]) + + # Expected-stream mismatch forces the server to return an error ack. + future = await js.publish_async("aerr", b"data", stream="WRONGSTREAM") + with pytest.raises(nats.js.errors.APIError): + await asyncio.wait_for(future, timeout=2) + + await nc.close() + @async_test async def test_publish_msg_ttl(self): """Test per-message TTL feature (requires NATS Server 2.11+)""" From ec65a71cc5881cee84d328d5d7916929724679e4 Mon Sep 17 00:00:00 2001 From: Casper Beyer Date: Tue, 1 Sep 2026 21:38:23 +0800 Subject: [PATCH 2/2] Resolve the ack future on any reply parsing failure json.loads and PubAck.from_response can both raise on an unexpected reply, and those escaped the subscription callback the same way the APIError did. The future's done callback is what pops the pending entry and releases the semaphore permit, so a stranded future also leaked a permit and left publish_async_completed() blocking forever. Route anything unexpected to the future, which also lets from_error raise straight into that handler. --- nats/src/nats/js/client.py | 18 ++++++++++-------- nats/tests/test_js.py | 35 ++++++++++++++++++++++++++++++++++- 2 files changed, 44 insertions(+), 9 deletions(-) diff --git a/nats/src/nats/js/client.py b/nats/src/nats/js/client.py index 43936e56..4e6c531e 100644 --- a/nats/src/nats/js/client.py +++ b/nats/src/nats/js/client.py @@ -165,19 +165,21 @@ async def _handle_async_reply(self, msg: Msg) -> None: try: resp = json.loads(msg.data) if "error" in resp: - # APIError.from_error raises rather than returning, so capture - # the constructed error and attach it to the future instead of - # letting it escape and leave the future pending forever. - try: - raise nats.js.errors.APIError.from_error(resp["error"]) - except nats.js.errors.APIError as err: - future.set_exception(err) - return + # Raises rather than returning; the handler below attaches the + # error to the future. + nats.js.errors.APIError.from_error(resp["error"]) ack = api.PubAck.from_response(resp) future.set_result(ack) except (asyncio.CancelledError, asyncio.InvalidStateError): pass + except Exception as err: + # Anything escaping here would strand the future: its done callback + # is what releases the semaphore permit and clears the pending + # entry, so publish_async_completed() would block forever and the + # permit would never come back. Resolve the future instead. + if not future.done(): + future.set_exception(err) async def publish( self, diff --git a/nats/tests/test_js.py b/nats/tests/test_js.py index 2694df61..9697f909 100644 --- a/nats/tests/test_js.py +++ b/nats/tests/test_js.py @@ -206,8 +206,41 @@ async def test_publish_async_error_ack_is_raised(self): # Expected-stream mismatch forces the server to return an error ack. future = await js.publish_async("aerr", b"data", stream="WRONGSTREAM") - with pytest.raises(nats.js.errors.APIError): + with pytest.raises(BadRequestError) as exc_info: await asyncio.wait_for(future, timeout=2) + self.assertEqual(exc_info.value.code, 400) + self.assertEqual(exc_info.value.err_code, 10060) + + # Resolving the future is also what releases the pending slot, so a + # regression strands the publish as well as swallowing the error. + await asyncio.wait_for(js.publish_async_completed(), timeout=2) + self.assertEqual(js.publish_async_pending(), 0) + + await nc.close() + + @async_test + async def test_publish_async_unparsable_ack_does_not_strand_future(self): + # An ack the client cannot parse must be routed to the future too. + # The future's done callback is what pops the pending entry and + # releases the semaphore permit, so letting the exception escape the + # subscription callback leaks a permit and makes + # publish_async_completed() block forever. + nc = NATS() + await nc.connect() + js = nc.jetstream(publish_async_max_pending=2) + await js.add_stream(name="APARSE", subjects=["aparse"]) + + with mock.patch.object(nats.js.api.PubAck, "from_response", side_effect=TypeError("bad ack")): + future = await js.publish_async("aparse", b"data") + with pytest.raises(TypeError): + await asyncio.wait_for(future, timeout=2) + + await asyncio.wait_for(js.publish_async_completed(), timeout=2) + self.assertEqual(js.publish_async_pending(), 0) + + # The permit came back, so further publishes still work. + ack = await asyncio.wait_for(await js.publish_async("aparse", b"data"), timeout=2) + self.assertEqual(ack.stream, "APARSE") await nc.close()