Raise publish_async ack errors on the returned future

nats/src/nats/js/client.py

@@ -165,14 +165,21 @@ 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)
-                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,

nats/tests/test_js.py

@@ -194,6 +194,56 @@ 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(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()
+
     @async_test
     async def test_publish_msg_ttl(self):
         """Test per-message TTL feature (requires NATS Server 2.11+)"""