If we time out on call (or initial connect) because we're disconnected, raiser ClientDisconnected.

src/ZEO/asyncio/client.py

@@ -679,7 +679,13 @@ class ClientRunner:
                 self.connected.set_result(None)
 
     def wait_for_result(self, future, timeout):
+        try:
             return future.result(self.timeout if timeout is False else timeout)
+        except concurrent.futures.TimeoutError:
+            if not self.client.ready:
+                raise ClientDisconnected("timed out waiting for connection")
+            else:
+                raise
 
     def call(self, method, *args, timeout=None):
         return self.__call(self.call_threadsafe, method, args)
@@ -781,7 +787,7 @@ class ClientThread(ClientRunner):
         if self.exception:
             raise self.exception
         if wait:
-            self.connected.result(self.timeout)
+            self.wait_for_result(self.connected, self.timeout)
 
     closed = False
     def close(self):

src/ZEO/asyncio/tests.py

@@ -28,7 +28,7 @@ class AsyncTests(setupstack.TestCase, ClientRunner):
 
         wrapper = mock.Mock()
         cache = MemoryCache()
-        self.set_options(addrs, wrapper, cache, 'TEST', read_only)
+        self.set_options(addrs, wrapper, cache, 'TEST', read_only, timeout=1)
 
         # We can also provide an event loop.  We'll use a testing loop
         # so we don't have to actually make any network connection.
@@ -604,6 +604,14 @@ class AsyncTests(setupstack.TestCase, ClientRunner):
         client.call_async_from_same_thread('foo', 1)
         self.assertEqual(self.parse(transport.pop()), (0, True, 'foo', (1, )))
 
+    def test_ClientDisconnected_on_call_timeout(self):
+        wrapper, cache, loop, client, protocol, transport, send, respond = (
+            self.start())
+        self.wait_for_result = super().wait_for_result
+        self.assertRaises(ClientDisconnected, self.call, 'foo')
+        client.ready = False
+        self.assertRaises(ClientDisconnected, self.call, 'foo')
+
     def unsized(self, data, unpickle=False):
         result = []
         while data: