fix(asyncio): fully disconnect removed cluster nodes in NodesManager.set_nodes

[eeshsaxena]

When removed from the cluster topology, old nodes only had their free (idle) connections cleaned up. In-use connections were left open indefinitely - the node is gone from nodes_cache, so acquire_connection() is never called again on it, meaning those sockets never get closed. This causes an "Unclosed ClusterNode object" warning and a real connection leak in long-running clients.

The fix calls disconnect() on each removed node instead of only disconnect_free_connections(). This closes all connections on removal, including ones still marked as in-use.

Note: in-flight commands on a removed node will fail with a connection error on this path, but that was already the case on reconnect attempts. The trade-off is correct - a node that left the topology should not keep sockets alive.

Test added in test_cluster.py verifies that both free and in-use connections are disconnected when a node is removed from the topology.

Fixes #4053

---

Note

Medium Risk
Changes cluster topology teardown behavior for removed nodes—active commands can fail sooner—but targets a real resource leak in long-lived clients with limited blast radius.

Overview
Fixes a connection leak when async cluster topology refresh drops a node: removed nodes no longer stay in memory with open sockets.

NodesManager.set_nodes (with remove_old=True) now schedules ClusterNode.disconnect() for each node that leaves the topology, closing free and in-use connections right away. The previous path only tore down idle connections and deferred in-use ones via reconnect marking, so sockets on evicted nodes could stay open indefinitely and trigger "Unclosed ClusterNode object" warnings.

In-flight work on a removed node may fail with a connection error when those sockets are closed; that matches the expectation that a node no longer in the cluster should not keep connections alive.

The cluster test was renamed and updated to assert immediate disconnect of both free and active connections.

^{Reviewed by Cursor Bugbot for commit 1b92aa9. Bugbot is set up for automated code reviews on this repo. Configure here.}

[chatgpt-codex-connector]

💡 Codex Review

Here are some automated review suggestions for this pull request.

Reviewed commit: 450fffa05c

ℹ️ About Codex in GitHub

Codex has been enabled to automatically review pull requests in this repo. Reviews are triggered when you

• Open a pull request for review
• Mark a draft as ready
• Comment "@codex review".

If Codex has suggestions, it will comment; otherwise it will react with 👍.

When you sign up for Codex through ChatGPT, Codex can also answer questions or update the PR, like "@codex address that feedback".

[chatgpt-codex-connector]

Let in-flight commands finish before disconnecting removed nodes

When a topology refresh removes a node after a mutating command has been sent but before its response is read, this immediate disconnect makes the command raise ConnectionError; RedisCluster.execute_command() then retries it on the refreshed topology by default. If the server already applied a command such as INCR or LPUSH, the retry applies it again and silently duplicates the write. Removed nodes should reject new work while marking active connections for disconnection only after their current operation completes.

AGENTS.md reference: AGENTS.md:L158-L162

Useful? React with 👍 / 👎.

[chatgpt-codex-connector]

Retrieve failures from detached disconnect tasks

If any connection's disconnect() times out or otherwise raises, ClusterNode.disconnect() re-raises that failure, but this detached task is only given a callback that discards it from the set without retrieving its exception. Once the task is released, asyncio reports Task exception was never retrieved through the application's loop exception handler. Add a completion callback that consumes and logs the exception, as other background disconnect sites in this package do.

Useful? React with 👍 / 👎.

[Sanjays2402]

ClusterNode.disconnect() gathers with return_exceptions=True and then re-raises the first exception, unlike disconnect_free_connections() which swallowed them. Since this is a fire-and-forget task whose only done_callback is discard, a node that goes away mid-topology-change (socket close raising) now turns into an unretrieved "Task exception was never retrieved" on the loop instead of a quiet cleanup. Probably worth wrapping the create_task target in a small helper that logs instead of raising.

[eeshsaxena]

Pushed a fix for the red asyncio jobs. The new test took default_host as a parameter, but in tests/test_asyncio/test_cluster.py that is a module-level constant rather than a fixture (unlike the sync suite, where tests/conftest.py does define a fixture by that name), so pytest errored at setup with fixture 'default_host' not found. Dropped the parameter - the constant resolves fine. Test passes locally.

[cursor]

Cursor Bugbot has reviewed your changes using default effort and found 1 potential issue.

^{Reviewed by Cursor Bugbot for commit 1b92aa9. Configure here.}

[cursor]

Unhandled exception in fire-and-forget disconnect task

Medium Severity

ClusterNode.disconnect() gathers connection disconnects with return_exceptions=True but then re-raises the first exception. Since this is a fire-and-forget task whose only done-callback is discard, any exception (e.g., a node disappearing mid-close) becomes an unhandled "Task exception was never retrieved" error on the event loop. The codebase already handles this correctly in _disconnect_and_release (which wraps disconnect() in try/except and logs), but the new code omits that protection.

 

^{Reviewed by Cursor Bugbot for commit 1b92aa9. Configure here.}

[Sanjays2402]

ClusterNode.disconnect() re-raises the first exception out of its gather, disconnect_free_connections() swallowed them (return_exceptions=True, no re-raise). This is a bare create_task whose only done_callback is the set discard, so nothing ever retrieves the result — a socket that errors on close now surfaces as "Task exception was never retrieved" on the loop instead of being ignored like before. Probably want to wrap it in a coroutine that logs and drops.