@@ -128,9 +128,13 @@ async def execute(self, command: MetaCommand, timeout: float | None) -> MetaResu
             with anyio.fail_after(timeout):
                 return await connection.execute_meta_command(command)
 
-    async def flush(self, delay: int) -> None:
+    async def flush(self, delay: int, timeout: float | None) -> None:
         async with self._borrow() as connection:
-            await connection.flush_all(delay)
+            if timeout is None:
+                await connection.flush_all(delay)
+                return
+            with anyio.fail_after(timeout):
+                await connection.flush_all(delay)
 
     async def close(self) -> None:
         self._closed = True
@@ -735,4 +739,4 @@ async def flush_all(self, delay: int = 0) -> None:
         positive("delay", delay)
         async with anyio.create_task_group() as tasks:
             for server in self._servers:
-                tasks.start_soon(server.flush, delay)
+                tasks.start_soon(server.flush, delay, self.default_timeout)
@@ -135,10 +135,13 @@ def __init__(
     @property
     def value(self) -> T:
         if self.status is not GetStatus.HIT:
+            # Chained so a FAILED result surfaces the ConnectionRefusedError or
+            # TimeoutError behind it. Without this the traceback reads like a
+            # usage mistake and the real cause is lost exactly when it matters.
             raise ResultValueError(
                 "key %r has no value to read: %s"
                 % (self.key, _NO_VALUE_REASONS.get(self.status, "the read failed"))
-            )
+            ) from self.error
         if self._value is _NO_VALUE:
             raise ResultValueError(
                 "key %r was a hit but the read did not ask for the value; "
 
@@ -237,3 +237,24 @@ async def test_batch_spanning_chunks_maps_every_response(client, monkeypatch):
     results = await client.batch([Get("achunk-%d" % i) for i in range(40)])
     assert [r.status for r in results] == [GetStatus.HIT] * 40
     assert [r.value for r in results] == ["v%d" % i for i in range(40)]
+
+
+@pytest.mark.asyncio
+async def test_flush_all_honours_the_client_timeout():
+    """flush_all used to have no deadline at all and could hang forever."""
+    import socket
+    import threading
+
+    listener = socket.socket()
+    listener.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
+    listener.bind(("127.0.0.1", 0))
+    listener.listen(4)
+    port = listener.getsockname()[1]
+    threading.Thread(
+        target=lambda: [listener.accept() for _ in range(4)], daemon=True
+    ).start()
+
+    silent = AsyncMetaClient(("127.0.0.1", port), timeout=0.3)
+    with pytest.raises(TimeoutError):
+        await silent.flush_all()
+    listener.close()
@@ -211,6 +211,17 @@ def send(self, commands, timeout=None):
     assert all(r.status is not MutationStatus.AMBIGUOUS for r in settled)
 
 
+def test_failed_result_value_chains_the_real_error():
+    """Reading .value on a FAILED result must not hide why it failed."""
+    client = MetaClient(("localhost", 1), timeout=0.3)
+    result = client.batch([Get("k")])[0]
+    assert isinstance(result.error, OSError)
+    with pytest.raises(ResultValueError) as info:
+        result.value
+    assert info.value.__cause__ is result.error
+    client.close()
+
+
 def test_server_failure_is_isolated_in_batch():
     client = MetaClient([("localhost", 11211), ("localhost", 1)], timeout=0.2)
     good = bad = None