From 0d644fce8f750ad0c90795b80203eb0d3fd558bb Mon Sep 17 00:00:00 2001
From: Lior Sventitzky <liorsve@amazon.com>
Date: Tue, 16 Sep 2025 11:58:24 +0300
Subject: [PATCH] Python: improve UDS socket error handling (#4733)

fixed uds error handling

Signed-off-by: Lior Sventitzky <liorsve@amazon.com>
---
 .../glide-async/python/glide/glide_client.py  |  7 ++++--
 python/tests/async_tests/test_async_client.py | 23 +++++++++++++++++++
 python/tests/test_api_consistency.py          |  1 +
 3 files changed, 29 insertions(+), 2 deletions(-)

diff --git a/python/glide-async/python/glide/glide_client.py b/python/glide-async/python/glide/glide_client.py
index 0b17fa9665..5e513a7d8c 100644
--- a/python/glide-async/python/glide/glide_client.py
+++ b/python/glide-async/python/glide/glide_client.py
@@ -338,7 +338,7 @@ async def _write_or_buffer_request(self, request: TRequest):
                     request.callback_idx if isinstance(request, CommandRequest) else 0
                 )
                 res_future = self._available_futures.pop(callback_idx, None)
-                if res_future:
+                if res_future and not res_future.done():
                     res_future.set_exception(e)
                 else:
                     ClientLogger.log(
@@ -355,7 +355,10 @@ async def _write_buffered_requests_to_socket(self) -> None:
         b_arr = bytearray()
         for request in requests:
             ProtobufCodec.encode_delimited(b_arr, request)
-        await self._stream.send(b_arr)
+        try:
+            await self._stream.send(b_arr)
+        except (anyio.ClosedResourceError, anyio.EndOfStream):
+            raise ClosingError("The communication layer was unexpectedly closed.")
 
     def _encode_arg(self, arg: TEncodable) -> bytes:
         """
diff --git a/python/tests/async_tests/test_async_client.py b/python/tests/async_tests/test_async_client.py
index 492b180a64..c5c59e2cca 100644
--- a/python/tests/async_tests/test_async_client.py
+++ b/python/tests/async_tests/test_async_client.py
@@ -412,6 +412,29 @@ async def connect_to_client():
         # Clean up the main client
         await client.close()
 
+    @pytest.mark.parametrize("cluster_mode", [True, False])
+    @pytest.mark.parametrize("protocol", [ProtocolVersion.RESP2, ProtocolVersion.RESP3])
+    async def test_UDS_socket_connection_failure(self, glide_client: TGlideClient):
+        """Test that the client's error handling during UDS socket connection failure"""
+        assert await glide_client.set("test_key", "test_value") == OK
+        assert await glide_client.get("test_key") == b"test_value"
+
+        # Force close the UDS connection to simulate socket failure
+        await glide_client._stream.aclose()
+
+        # Verify a ClosingError is raised
+        with pytest.raises(
+            ClosingError, match="The communication layer was unexpectedly closed"
+        ):
+            await glide_client.get("test_key")
+
+        # Verify the client is closed
+        with pytest.raises(
+            ClosingError,
+            match="Unable to execute requests; the client is closed. Please create a new client.",
+        ):
+            await glide_client.get("test_key")
+
 
 @pytest.mark.anyio
 class TestCommands:
diff --git a/python/tests/test_api_consistency.py b/python/tests/test_api_consistency.py
index 3bea180a71..4fd3fc26b5 100644
--- a/python/tests/test_api_consistency.py
+++ b/python/tests/test_api_consistency.py
@@ -69,6 +69,7 @@
         "test_inflight_request_limit",
         "test_statistics",
         "test_select",
+        "test_UDS_socket_connection_failure",
     ],
     "sync_only": ["test_sync_fork"],
 }