Skip to content

Raise publish_async ack errors on the returned future #994

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
caspervonb wants to merge 1 commit into
base: main
Choose a base branch
from
Open

Raise publish_async ack errors on the returned future #994

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
9 changes: 7 additions & 2 deletions nats/src/nats/js/client.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -165,8 +165,13 @@ async def _handle_async_reply(self, msg: Msg) -> None:
try:
resp = json.loads(msg.data)

@renerocksai renerocksai Aug 4, 2026

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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.

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"])
Comment thread
caspervonb marked this conversation as resolved.
except nats.js.errors.APIError as err:
future.set_exception(err)
return

ack = api.PubAck.from_response(resp)
Expand Down
17 changes: 17 additions & 0 deletions nats/tests/test_js.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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):

@renerocksai renerocksai Aug 4, 2026

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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.

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+)"""
Expand Down
Loading