Raise publish_async ack errors on the returned future

[caspervonb]

APIError.from_error raises rather than returns, so the error branch in _handle_async_reply threw before set_exception; the error escaped into the delivery task and the ack future was left pending forever. Capture and attach it to the future.

Fixes #985.

[synadia-claude-reviewer]

Claude finished @caspervonb's task in 1m 37s —— View job

---

PR Review

• Gather context
• Read PR diff and relevant source files
• Analyze correctness, asyncio patterns, and test coverage
• Post review

---

Verdict: Correct fix, one minor nit.

The root cause is clear and the description is accurate: APIError.from_error always raises (every branch in errors.py:80-91 is a raise), so the original call err = APIError.from_error(...) threw before set_exception was ever reached. The exception escaped the delivery callback and the ack future was left in a permanently pending state.

What the fix does right:

• Wrapping the call in a try/except APIError cleanly captures the raised exception and routes it to future.set_exception(err), so callers awaiting the future see the error rather than hanging forever.
• The outer except (asyncio.CancelledError, asyncio.InvalidStateError) still guards future.set_exception(err) in case the future is already completed — consistent with the success path.
• The return after the inner block is correctly placed at the if "error" level, not inside the inner try/except.

One nit (non-blocking): The raise keyword before nats.js.errors.APIError.from_error(...) is redundant — since from_error raises internally, the outer raise is never reached. Left an inline comment with a one-line suggestion to drop it.

Test: The regression test is well-constructed — triggering a server-side error via stream="WRONGSTREAM" mismatch and using asyncio.wait_for(..., timeout=2) means a regression would produce a TimeoutError rather than silently passing.

[renerocksai]

Please bear in mind, I am fairly new to nats-py and might be wrong in my assessments. The review focused on idiomatic / good-quality async Python code; I haven't evaluated NATS-specific design choices.

1. I think the same "future left pending forever" hole is still open on the non-error branch of _handle_async_reply — the fallback handler only catches CancelledError/InvalidStateError, so a JSONDecodeError out of json.loads or a TypeError out of PubAck.from_response escapes exactly the way the APIError did. Details inline.

2. The root cause looks like APIError.from_error being an unannotated always-raising factory, sitting right under from_msg(...) -> NoReturn. No inline for this one since nats/src/nats/js/errors.py isn't in the diff: annotating from_error as -> NoReturn (errors.py:80) would turn err = APIError.from_error(...) into something a type checker can flag, rather than a runtime surprise that needs a comment to explain. The larger alternative — having from_error return the error and updating the three raise APIError.from_error(...) call sites — would drop the try/except round-trip and the comment entirely, but it changes public behaviour, so quite possibly out of scope for a fix PR.

3. The rest are small test tightenings — hedged inline, take or leave.

[renerocksai]

I think this try block still leaves the non-error branch exposed to the very bug this PR fixes. The handler at the bottom catches only asyncio.CancelledError / asyncio.InvalidStateError, but two statements inside it can raise something else:

• json.loads(msg.data) → json.JSONDecodeError (a ValueError) on a payload that isn't JSON
• api.PubAck.from_response(resp) on line 177 → TypeError, since Base.from_response ends in cls(**params) and PubAck.stream / PubAck.seq have no defaults

Either one propagates out of the subscription callback into Subscription._wait_for_msgs, which hands it to error_cb and carries on — so the future is never resolved, handle_done never runs, the token stays in _publish_async_futures, the semaphore permit is never released, and publish_async_completed() blocks forever. Same failure mode as #985, just reached down a different branch.

Since it's the same defect class, it might be worth having the fallback route to the future as well:

except (asyncio.CancelledError, asyncio.InvalidStateError):
    pass
except Exception as err:
    if not future.done():
        future.set_exception(err)

That keeps the current behaviour for the two existing cases and stops anything else from stranding the future.

[renerocksai]

Two small tightenings, both hedged:

1. pytest.raises(nats.js.errors.APIError) pins only the base class that every JS API error inherits from. The point of the fix is that the specific error, with its fields intact, now reaches the caller — so asserting that would be stronger, e.g. with pytest.raises(...) as exc_info: plus a check on exc_info.value.err_code / .code (or a match= on the description). If an expected-stream mismatch comes back as a 400, BadRequestError would be the concrete type.

2. The other half of the regression isn't covered. Before the fix the future never completed, so handle_done never ran: publish_async_pending() stayed at 1 and publish_async_completed() would hang. test_publish_async just above already asserts exactly that, so mirroring it here would lock in the leak side too:

await asyncio.wait_for(js.publish_async_completed(), timeout=2)
self.assertEqual(js.publish_async_pending(), 0)

(Wrapped in wait_for so a regression fails rather than hangs.)

Very minor: the file does from nats.js.errors import * and the neighbouring tests use the bare names (NoStreamResponseError, TooManyStalledMsgsError), so an unqualified APIError here would match the surrounding style.