[Backport 2.1] Python: improve UDS socket error handling (#4733)

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:
         """

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:

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"],
 }