fix(async): include event loop ID in connection alias to prevent reusing closed connections (#3086)

# Description

This PR fixes a critical issue where `AsyncMilvusClient` could reuse a
connection associated with a closed event loop, causing `RuntimeError:
Event loop is closed` when running multiple async tasks or tests
sequentially.

## Problem

* The `create_connection` utility generates a connection alias based on
URI, user, and other parameters but **ignores the running event loop**.
* In `AsyncMilvusClient`, connections are cached globally by this alias.
* If an event loop is closed (e.g., after `asyncio.run()` finishes) and
a new loop is started, the client reuses the **cached connection** which
is bound to the **old, closed** loop.
* This results in errors like: `RuntimeError: Event loop is closed` when
trying to use the client in the new loop.

## Solution

* Updated `create_connection` in `pymilvus/milvus_client/_utils.py` to
include the **current event loop ID** in the connection alias when
`use_async=True`.
* **New Alias Format:** `async-{uri}-{hash}-loop{loop_id}`
* This ensures that each event loop gets its own dedicated connection
instance.
* Added a check to raise `RuntimeError` if no running loop is found in
an async context (fail-fast instead of silent failure).

## Verification

* Added a new unit test `tests/test_async_milvus_client_reuse.py` to
reproduce the issue and verify the fix.
* The test confirms that different event loops generate different
connection aliases, preventing invalid reuse.
* Existing tests pass.

## Related Issue

#3087

Co-authored-by: silas.jiang <silas.jiang@zilliz.com>

Author: jac0626

pymilvus/milvus_client/_utils.py

@@ -1,3 +1,4 @@
+import asyncio
 import hashlib
 import logging
 
@@ -33,8 +34,15 @@ def create_connection(
             md5.update(token.encode())
             auth_fmt = f"{md5.hexdigest()}"
 
-        # different uri, auth, db_name cannot share the same connection
-        not_empty = [v for v in [use_async_fmt, uri, db_name, auth_fmt] if v]
+        # For async connections, include event loop ID in alias to prevent
+        # reusing connections from closed event loops
+        loop_id_fmt = ""
+        if use_async:
+            loop = asyncio.get_running_loop()
+            loop_id_fmt = f"loop{id(loop)}"
+
+        # different uri, auth, db_name, and event loop (for async) cannot share the same connection
+        not_empty = [v for v in [use_async_fmt, uri, db_name, auth_fmt, loop_id_fmt] if v]
         using = "-".join(not_empty)
 
     if connections.has_connection(using):

tests/test_async_milvus_client_reuse.py

@@ -0,0 +1,48 @@
+import asyncio
+import pytest
+from unittest.mock import patch, MagicMock
+from pymilvus import AsyncMilvusClient
+
+class TestAsyncConnectionReuse:
+    def test_async_client_alias_different_loops(self):
+        """
+        Test that AsyncMilvusClient generates different connection aliases
+        for different event loops to avoid reusing closed connections.
+        """
+        uri = "http://localhost:19530"
+        
+        async def create_client():
+            # Mock connections.connect to avoid real network connection
+            # Mock utility.get_server_type to avoid checking server version
+            with patch("pymilvus.orm.connections.Connections.connect") as mock_connect, \
+                 patch("pymilvus.orm.utility.get_server_type", return_value="milvus"):
+                
+                client = AsyncMilvusClient(uri=uri)
+                # Return the alias used and the current loop id
+                return client._using, id(asyncio.get_running_loop())
+
+        # Manually create loops to ensure we have distinct objects
+        loop1 = asyncio.new_event_loop()
+        loop2 = asyncio.new_event_loop()
+        
+        try:
+            # Run in loop 1
+            alias1, loop1_id = loop1.run_until_complete(create_client())
+            
+            # Run in loop 2
+            alias2, loop2_id = loop2.run_until_complete(create_client())
+            
+            print(f"Loop 1 ID: {loop1_id}, Alias 1: {alias1}")
+            print(f"Loop 2 ID: {loop2_id}, Alias 2: {alias2}")
+
+            # Since loop1 and loop2 are both alive (referenced), they must have different IDs
+            assert loop1_id != loop2_id, "Two active loops must have different IDs"
+            assert alias1 != alias2, "Aliases should be different for different loops"
+            
+            # Check if loop ID is part of the alias
+            assert f"loop{loop1_id}" in alias1
+            assert f"loop{loop2_id}" in alias2
+            
+        finally:
+            loop1.close()
+            loop2.close()