Add MaxReconnectAttemptsExceededError (#735)

[negativeOne1]

Introduces a new MaxReconnectAttemptsExceededError that is raised when the NATS client exhausts all reconnection attempts across all servers in the pool. This allows applications to
distinguish between temporary connection issues and permanent failures where all retry limits have been exceeded.

[Copilot]

Pull Request Overview

Adds a dedicated MaxReconnectAttemptsExceededError to signal when the client has exhausted all reconnection attempts across all servers, enabling applications to differentiate permanent failures from transient disconnects.

• Introduces MaxReconnectAttemptsExceededError in nats.errors
• Raises this error from client._select_next_server when all servers exceed max_reconnect_attempts
• Adds a test validating the new error propagation and state

Reviewed Changes

Copilot reviewed 3 out of 3 changed files in this pull request and generated 5 comments.

File	Description
tests/test_client.py	New test to validate error raised and callbacks/state when all reconnect attempts across servers are exhausted
nats/errors.py	Adds MaxReconnectAttemptsExceededError with a descriptive str and max_attempts attribute
nats/aio/client.py	Emits MaxReconnectAttemptsExceededError when no servers remain within allowed reconnect attempts

---

_{Tip: Customize your code reviews with copilot-instructions.md. Create the file or learn how to get started.}

[Copilot]

[nitpick] Use Optional[int] instead of the PEP 604 union (int | None) for broader Python compatibility and to align with typical typing style in mixed-version codebases. Suggested change: def init(self, max_attempts: Optional[int] = None) -> None (ensure Optional is imported from typing).

[Copilot]

[nitpick] Add a short class docstring describing when this error is raised and the meaning of max_attempts. This improves discoverability and clarifies the API contract for consumers.

[Copilot]

The executor futures are not awaited, which can lead to nondeterministic behavior and test flakiness. Collect and await the futures to ensure all servers are stopped before proceeding, e.g., stops = [asyncio.get_running_loop().run_in_executor(None, s.stop) for s in self.server_pool]; await asyncio.gather(*stops).

Suggested change

	for server in self.server_pool:
	asyncio.get_running_loop().run_in_executor(None, server.stop)
	stops = [asyncio.get_running_loop().run_in_executor(None, server.stop) for server in self.server_pool]
	await asyncio.gather(*stops)

[Copilot]

Using a fixed sleep makes the test timing-dependent and potentially flaky. Replace with a condition-based wait (polling nc.is_closed or the error callback capturing MaxReconnectAttemptsExceededError with a timeout) to deterministically wait until the client has exhausted reconnects.

Suggested change

	# Wait for the client to exhaust all reconnection attempts
	await asyncio.sleep(1.0)
	# Wait for the client to exhaust all reconnection attempts (deterministically)
	timeout = 2.0 # seconds
	interval = 0.05 # seconds
	waited = 0.0
	while not nc.is_closed and not any(isinstance(e, nats.errors.MaxReconnectAttemptsExceededError) for e in errors):
	await asyncio.sleep(interval)
	waited += interval
	if waited >= timeout:
	break

[Copilot]

[nitpick] The explicit len(self._server_pool) == 0 check is redundant since all([]) evaluates to True; the condition can be simplified to just all(...). This reduces duplication and keeps the intent clear.

Suggested change

	if len(self._server_pool) == 0 or all(
	if all(