fix(vrtManager): libvirt connections open within a time limit

libvirt has no timeout for opening a connection. Measured in the test
setup: an unreachable host blocked qemu+tcp for 135 s and qemu+ssh for
133 s; a daemon that accepts and never answers blocks forever. Gunicorn
runs sync workers and the instance list connects to every compute, so
one unreachable compute held a worker for minutes per page load.

- A connection is opened in a helper thread, at most one per
  connection at a time. The request waits LIBVIRT_CONNECT_TIMEOUT
  (default 5 s). An open nobody waits for any more keeps running; until
  it ends, requests get its error at once, and if it succeeds later the
  connection is used. A failed open is not repeated for
  LIBVIRT_RETRY_AFTER (default 30 s). A stuck host holds one thread and
  does not keep other hosts from connecting.
- qemu+ssh runs ssh through vrtManager/libvirt-ssh (the URI's command
  parameter), which adds ConnectTimeout=10 and BatchMode=yes, so an
  abandoned ssh process ends and no prompt can hang. The user's ssh
  configuration (aliases, ProxyJump, keys) still applies.
- The close callback acts only for the connection it was registered
  for: a late callback of a replaced connection dropped its successor.
  The reason is reported as text. A connection that closed before it
  was published is an error.
- Compute.status is a bool (it returned the exception object, so the
  "host is down" check in refresh_instance_database never fired); the
  message is Compute.connection_error.

Unit tests cover the timeout, retry delay, late success, single open,
a stuck host next to a healthy one and the close callback. Live tests
(vrtManager/test_live_connection.py), for TCP and SSH computes of the
test setup: an unreachable address (no ssh left after 10 s), an endpoint
that never answers, a slow open completing later, a host that goes
silent on an established connection (keepalive ends calls within 45 s,
the next request reconnects), an ssh_config alias and a ProxyJump host.

computes/models.py

@@ -17,12 +17,14 @@ class Compute(Model):
 
     @cached_property
     def status(self):
-        # return connection_manager.host_is_up(self.type, self.hostname)
-        # TODO: looks like socket has problems connecting via VPN
-        if isinstance(self.connection, virConnect):
-            return True
-        else:
-            return self.connection
+        """True if WebVirtCloud can connect to the host (within the connect
+        timeout; a host that failed recently is not tried again at once)."""
+        return isinstance(self.connection, virConnect)
+
+    @cached_property
+    def connection_error(self):
+        """Why the host cannot be reached, or None."""
+        return None if self.status else str(self.connection)
 
     @cached_property
     def connection(self):

computes/test_ownership.py

@@ -34,6 +34,7 @@ class OwnershipTestCase(TestCase):
     def compute(self, name):
         compute = Compute.objects.create(name=name, hostname=f"{name}.example", login="", password="", type=1)
         compute.__dict__["proxy"] = MagicMock()
+        compute.__dict__["status"] = True  # reachable
         self.on(compute, [])
         return compute
 

computes/test_reconciliation.py

@@ -40,6 +40,7 @@ class InstanceReconciliationTestCase(TestCase):
         mock_proxy = MagicMock()
         mock_proxy.wvm.listAllDomains.return_value = [dom1, dom2]
         self.compute.__dict__["proxy"] = mock_proxy
+        self.compute.__dict__["status"] = True  # reachable
 
         refresh_instance_database(self.compute)
 
@@ -61,6 +62,7 @@ class InstanceReconciliationTestCase(TestCase):
         mock_proxy = MagicMock()
         mock_proxy.wvm.listAllDomains.return_value = []
         self.compute.__dict__["proxy"] = mock_proxy
+        self.compute.__dict__["status"] = True  # reachable
 
         refresh_instance_database(self.compute)
 
@@ -82,6 +84,7 @@ class InstanceReconciliationTestCase(TestCase):
         mock_proxy = MagicMock()
         mock_proxy.wvm.listAllDomains.return_value = [dom_renamed]
         self.compute.__dict__["proxy"] = mock_proxy
+        self.compute.__dict__["status"] = True  # reachable
 
         refresh_instance_database(self.compute)
 
@@ -105,6 +108,7 @@ class InstanceReconciliationTestCase(TestCase):
         mock_proxy = MagicMock()
         mock_proxy.wvm.listAllDomains.side_effect = Exception("Libvirt socket connection refused")
         self.compute.__dict__["proxy"] = mock_proxy
+        self.compute.__dict__["status"] = True  # reachable
 
         refresh_instance_database(self.compute)
 
@@ -124,6 +128,7 @@ class InstanceReconciliationTestCase(TestCase):
         mock_proxy = MagicMock()
         mock_proxy.wvm.listAllDomains.return_value = [broken_dom]
         self.compute.__dict__["proxy"] = mock_proxy
+        self.compute.__dict__["status"] = True  # reachable
 
         refresh_instance_database(self.compute)
 

computes/tests.py

@@ -214,6 +214,7 @@ class ComputeConcurrencyTestCase(TransactionTestCase):
         mock_proxy = MagicMock()
         mock_proxy.wvm.listAllDomains.return_value = [mock_dom1, mock_dom2]
         compute.proxy = mock_proxy
+        compute.status = True  # reachable
 
         exceptions = []
 
@@ -253,6 +254,7 @@ class ComputeConcurrencyTestCase(TransactionTestCase):
         mock_proxy = MagicMock()
         mock_proxy.wvm.listAllDomains.return_value = []
         compute.proxy = mock_proxy
+        compute.status = True  # reachable
 
         # 1. When lock directory cannot be acquired
         with patch("computes.utils._get_lock_directory", return_value=None):

instances/templates/allinstances.html

@@ -18,7 +18,7 @@
     {% for compute in computes %}
         {% if compute.status is not True %}
             <div class="alert alert-danger alert-dismissible fade show" role="alert">
-                {% trans 'Problem occurred with host' %} {{ compute.name }}: {{ compute.status }}
+                {% trans 'Problem occurred with host' %} {{ compute.name }}: {{ compute.connection_error }}
                 <button type="button" class="btn-close" data-bs-dismiss="modal" aria-label="Close"></button>
             </div>
         {% endif %}

vrtManager/connection.py

@@ -1,6 +1,9 @@
+import os
 import re
 import socket
 import threading
+import time
+from urllib.parse import quote
 
 import libvirt
 from django.conf import settings
@@ -16,6 +19,20 @@ TLS_PORT = 16514
 SSH_PORT = 22
 TCP_PORT = 16509
 
+# libvirt has no timeout for opening a connection: an unreachable host
+# blocks for the kernel's TCP connect timeout (about 2 minutes), a daemon
+# that accepts but never answers blocks forever, and such an open cannot be
+# cancelled. Connections are opened in a helper thread, at most one per
+# connection at a time, so a stuck host holds one thread and nothing else;
+# the caller waits at most CONNECT_TIMEOUT seconds, and a host that failed is
+# not tried again for RETRY_AFTER seconds.
+CONNECT_TIMEOUT = getattr(settings, "LIBVIRT_CONNECT_TIMEOUT", 5)
+RETRY_AFTER = getattr(settings, "LIBVIRT_RETRY_AFTER", 30)
+CLOSE_REASONS = {0: "error", 1: "end of file", 2: "keepalive timeout", 3: "client closed"}
+# ssh with a connect timeout for qemu+ssh (libvirt runs the "command" binary
+# in place of ssh and never passes one itself)
+SSH_COMMAND = os.path.join(os.path.dirname(os.path.abspath(__file__)), "libvirt-ssh")
+
 
 class wvmEventLoop(threading.Thread):
     """ Event Loop Class"""
@@ -58,6 +75,8 @@ class wvmConnection(object):
         self.connection_state_lock = threading.Lock()
         self.connection = None
         self.last_error = None
+        self._opening = None  # threading.Event of the open in progress
+        self._retry_at = 0.0  # monotonic time before which no new open is tried
 
         # credentials
         self.host = host
@@ -66,39 +85,70 @@ class wvmConnection(object):
         self.type = conn
 
     def connect(self):
-        self.connection_state_lock.acquire()
-        try:
-            # recheck if we have a connection (it may have been
-            if not self.connected:
-                if self.type == CONN_TCP:
-                    self.__connect_tcp()
-                elif self.type == CONN_SSH:
-                    self.__connect_ssh()
-                elif self.type == CONN_TLS:
-                    self.__connect_tls()
-                elif self.type == CONN_SOCKET:
-                    self.__connect_socket()
-                else:
-                    raise ValueError(f'"{self.type}" is not a valid connection type')
+        """Open the connection if needed, waiting at most CONNECT_TIMEOUT.
+        On failure last_error says why and connected stays False."""
+        with self.connection_state_lock:
+            if self.connected:
+                return
+            if self._opening is not None and self._opening.abandoned:
+                # A caller already waited for this open in vain; nobody waits
+                # again until it ends (one thread per connection, however long).
+                return
+            if time.monotonic() < self._retry_at:
+                return  # failed recently: answer with the last error at once
+            if self._opening is None:
+                self._opening = threading.Event()
+                self._opening.abandoned = False
+                threading.Thread(target=self._open, args=(self._opening,), daemon=True).start()
+            opening = self._opening
 
-                if self.connected:
-                    # do some preprocessing of the connection:
-                    #     * set keep alive interval
-                    #     * set connection close/fail handler
-                    try:
-                        self.connection.setKeepAlive(
-                            connection_manager.keepalive_interval, connection_manager.keepalive_count
-                        )
-                        try:
-                            self.connection.registerCloseCallback(self.__connection_close_callback, None)
-                        except Exception:
-                            # Temporary fix for libvirt > libvirt-0.10.2-41
-                            pass
-                    except libvirtError as e:
-                        # hypervisor driver does not seem to support persistent connections
-                        self.last_error = str(e)
+        if not opening.wait(CONNECT_TIMEOUT):
+            with self.connection_state_lock:
+                if self._opening is opening:  # still running: give up waiting, not the open
+                    self.last_error = f"Connection Failed: {self.host} did not answer within {CONNECT_TIMEOUT} s"
+                    opening.abandoned = True
+
+    def _open(self, opening):
+        """Helper thread: open, configure and publish the connection."""
+        connection = error = None
+        try:
+            connection = self._open_transport()
+            if not connection.isAlive():  # closed again before it was published
+                raise util.OperationError("the connection closed right after opening")
+            # Drivers without keepalive or close callbacks work without them
+            try:
+                connection.setKeepAlive(connection_manager.keepalive_interval, connection_manager.keepalive_count)
+            except libvirtError:
+                pass
+            try:
+                connection.registerCloseCallback(self.__connection_close_callback, None)
+            except libvirtError:
+                pass
+        except Exception as e:  # libvirtError, or anything else from the transport
+            error = f"Connection Failed: {str(e)}"
         finally:
-            self.connection_state_lock.release()
+            with self.connection_state_lock:
+                if error is None:
+                    self.connection = connection
+                    self.last_error = None
+                    self._retry_at = 0.0
+                else:
+                    self.connection = None
+                    self.last_error = error
+                    self._retry_at = time.monotonic() + RETRY_AFTER
+                self._opening = None
+            opening.set()
+
+    def _open_transport(self):
+        if self.type == CONN_TCP:
+            return self.__connect_tcp()
+        if self.type == CONN_SSH:
+            return self.__connect_ssh()
+        if self.type == CONN_TLS:
+            return self.__connect_tls()
+        if self.type == CONN_SOCKET:
+            return self.__connect_socket()
+        raise ValueError(f'"{self.type}" is not a valid connection type')
 
     @property
     def connected(self):
@@ -121,65 +171,30 @@ class wvmConnection(object):
         return 0
 
     def __connection_close_callback(self, connection, reason, opaque=None):
-        self.connection_state_lock.acquire()
-        try:
-            # on server shutdown libvirt module gets freed before the close callbacks are called
-            # so we just check here if it is still present
-            if libvirt is not None:
-                self.last_error = reason
-
+        with self.connection_state_lock:
+            # A late callback of a connection that was already replaced must
+            # not drop its successor.
+            if connection is not self.connection:
+                return
+            self.last_error = f"Connection closed ({CLOSE_REASONS.get(reason, reason)})"
             # prevent other threads from using the connection (in the future)
             self.connection = None
-        finally:
-            self.connection_state_lock.release()
 
     def __connect_tcp(self):
         flags = [libvirt.VIR_CRED_AUTHNAME, libvirt.VIR_CRED_PASSPHRASE]
         auth = [flags, self.__libvirt_auth_credentials_callback, None]
-        uri = f"qemu+tcp://{self.host}/system"
-
-        try:
-            self.connection = libvirt.openAuth(uri, auth, 0)
-            self.last_error = None
-
-        except libvirtError as e:
-            self.last_error = f"Connection Failed: {str(e)}"
-            self.connection = None
+        return libvirt.openAuth(f"qemu+tcp://{self.host}/system", auth, 0)
 
     def __connect_ssh(self):
-        uri = "qemu+ssh://%s@%s/system" % (self.login, self.host)
-
-        try:
-            self.connection = libvirt.open(uri)
-            self.last_error = None
-
-        except libvirtError as e:
-            self.last_error = f"Connection Failed: {str(e)} --- " + repr(libvirt.virGetLastError())
-            self.connection = None
+        return libvirt.open(f"qemu+ssh://{self.login}@{self.host}/system?command={quote(SSH_COMMAND)}")
 
     def __connect_tls(self):
         flags = [libvirt.VIR_CRED_AUTHNAME, libvirt.VIR_CRED_PASSPHRASE]
         auth = [flags, self.__libvirt_auth_credentials_callback, None]
-        uri = "qemu+tls://%s@%s/system" % (self.login, self.host)
-
-        try:
-            self.connection = libvirt.openAuth(uri, auth, 0)
-            self.last_error = None
-
-        except libvirtError as e:
-            self.last_error = f"Connection Failed: {str(e)}"
-            self.connection = None
+        return libvirt.openAuth(f"qemu+tls://{self.login}@{self.host}/system", auth, 0)
 
     def __connect_socket(self):
-        uri = "qemu:///system"
-
-        try:
-            self.connection = libvirt.open(uri)
-            self.last_error = None
-
-        except libvirtError as e:
-            self.last_error = f"Connection Failed: {str(e)}"
-            self.connection = None
+        return libvirt.open("qemu:///system")
 
     def close(self):
         """
@@ -286,7 +301,7 @@ class wvmConnectionManager(object):
                 self._connections_lock.release()
 
         if not connection.connected:
-            # (re-)connect; the connection's own lock serializes concurrent attempts
+            # (re-)connect within CONNECT_TIMEOUT; one open per connection at a time
             connection.connect()
 
         if connection.connected:

vrtManager/libvirt-ssh

@@ -0,0 +1,12 @@
+#!/bin/sh
+# ssh for libvirt qemu+ssh connections (the URI's "command" parameter).
+# libvirt passes no connect timeout, so an unreachable host would keep this
+# ssh process waiting for the kernel's TCP timeout. BatchMode makes ssh fail
+# instead of waiting for a password or host key prompt nobody can answer.
+# The timeout is fixed: libvirt passes this script no environment or extra
+# arguments. It bounds only ssh's TCP connect; WebVirtCloud stops waiting
+# after LIBVIRT_CONNECT_TIMEOUT on its own, so an abandoned ssh lives at most
+# this long.
+# Everything else comes from the user's ssh configuration (aliases,
+# ProxyJump, keys).
+exec ssh -o ConnectTimeout=10 -o BatchMode=yes "$@"

vrtManager/test_connection.py

@@ -1,17 +1,19 @@
 """Reachability probe and connections that do not block each other."""
 
+import os
 import socket
 import threading
 import time
 import unittest
-from unittest.mock import patch
+from unittest.mock import MagicMock, patch
 
 from django.conf import settings
 
 if not settings.configured:
     settings.configure(MAC_OUI="52:54:10")
 
-from vrtManager.connection import CONN_TCP, wvmConnectionManager
+import libvirt
+from vrtManager.connection import CONN_SSH, CONN_TCP, SSH_COMMAND, wvmConnectionManager
 
 
 def closed_port():
@@ -61,5 +63,148 @@ class ConnectionManagerTestCase(unittest.TestCase):
             slow.join()
 
 
+
+def alive_connection():
+    conn = MagicMock()
+    conn.isAlive.return_value = 1
+    return conn
+
+
+class BoundedOpenTestCase(unittest.TestCase):
+    """Opening waits at most CONNECT_TIMEOUT; a failed host is not retried
+    for RETRY_AFTER; one open per connection at a time."""
+
+    def setUp(self):
+        for name, value in (("wvmEventLoop", MagicMock()), ("CONNECT_TIMEOUT", 0.3), ("RETRY_AFTER", 1.0)):
+            patcher = patch(f"vrtManager.connection.{name}", value)
+            patcher.start()
+            self.addCleanup(patcher.stop)
+        self.manager = wvmConnectionManager()
+        self.release = threading.Event()
+        self.addCleanup(self.release.set)
+        self.calls = 0
+
+    def open_auth(self, result):
+        def open_auth(uri, auth, flags):
+            self.calls += 1
+            self.release.wait(5)
+            if isinstance(result, Exception):
+                raise result
+            return result
+        return patch("vrtManager.connection.libvirt.openAuth", side_effect=open_auth)
+
+    def get(self, host="h"):
+        started = time.monotonic()
+        try:
+            return self.manager.get_connection(host, "u", "p", CONN_TCP), time.monotonic() - started
+        except libvirt.libvirtError as err:
+            return err, time.monotonic() - started
+
+    def test_a_hanging_open_is_given_up_after_the_timeout(self):
+        with self.open_auth(alive_connection()):
+            result, took = self.get()
+        self.assertIsInstance(result, libvirt.libvirtError)
+        self.assertIn("did not answer within", str(result))
+        self.assertLess(took, 1.0)
+
+    def test_within_retry_after_the_last_error_comes_at_once(self):
+        with self.open_auth(alive_connection()):
+            self.get()
+            result, took = self.get()
+        self.assertIn("did not answer within", str(result))
+        self.assertLess(took, 0.1)
+        self.assertEqual(self.calls, 1)
+
+    def test_a_late_success_is_used(self):
+        conn = alive_connection()
+        with self.open_auth(conn):
+            self.get()
+            self.release.set()
+            time.sleep(0.1)
+            result, took = self.get()
+        self.assertIs(result, conn)
+        self.assertEqual(self.calls, 1)
+
+    def test_a_failed_host_is_tried_again_after_retry_after(self):
+        self.release.set()
+        with self.open_auth(libvirt.libvirtError("refused")):
+            self.get()
+            self.get()
+            self.assertEqual(self.calls, 1)
+            time.sleep(1.1)
+            self.get()
+        self.assertEqual(self.calls, 2)
+
+    def test_concurrent_requests_share_one_open(self):
+        with self.open_auth(alive_connection()):
+            threads = [threading.Thread(target=self.get) for _ in range(5)]
+            for t in threads:
+                t.start()
+            for t in threads:
+                t.join()
+        self.assertEqual(self.calls, 1)
+
+    def test_a_stuck_host_does_not_keep_other_hosts_from_connecting(self):
+        healthy = alive_connection()
+
+        def open_auth(uri, auth, flags):
+            if "stuck" in uri:
+                self.release.wait(5)  # an open that never returns
+            return healthy
+
+        with patch("vrtManager.connection.libvirt.openAuth", side_effect=open_auth):
+            for i in range(12):
+                self.get(f"stuck{i}")
+            result, took = self.get("healthy")
+        self.assertIs(result, healthy)
+
+    def test_an_abandoned_open_is_not_waited_for_again_nor_duplicated(self):
+        conn = alive_connection()
+        with self.open_auth(conn):
+            self.get()
+            time.sleep(1.1)  # past RETRY_AFTER, the open still runs
+            result, took = self.get()
+            self.assertIsInstance(result, libvirt.libvirtError)
+            self.assertLess(took, 0.1)
+            self.release.set()  # the host answers at last
+            time.sleep(0.1)
+            result, _ = self.get()
+        self.assertIs(result, conn)
+        self.assertEqual(self.calls, 1)
+
+    def test_a_late_close_callback_does_not_drop_the_new_connection(self):
+        self.release.set()
+        old, new = alive_connection(), alive_connection()
+        with self.open_auth(old):
+            self.get()
+        entry = self.manager._search_connection("h", "u", "p", CONN_TCP)
+        old.isAlive.return_value = 0  # seen dead, reconnected
+        with self.open_auth(new):
+            self.assertIs(self.get()[0], new)
+        entry._wvmConnection__connection_close_callback(old, 1)
+        self.assertIs(self.get()[0], new)
+        entry._wvmConnection__connection_close_callback(new, 2)
+        self.assertIsNone(entry.connection)
+        self.assertEqual(entry.last_error, "Connection closed (keepalive timeout)")
+
+    def test_a_connection_closed_before_it_is_published_is_an_error(self):
+        self.release.set()
+        dead = alive_connection()
+        dead.isAlive.return_value = 0
+        with self.open_auth(dead):
+            result, _ = self.get()
+        self.assertIn("closed right after opening", str(result))
+
+    def test_ssh_runs_ssh_with_a_connect_timeout(self):
+        self.release.set()
+        with patch("vrtManager.connection.libvirt.open", return_value=alive_connection()) as libvirt_open:
+            self.manager.get_connection("host", "root", "", CONN_SSH)
+        uri = libvirt_open.call_args[0][0]
+        self.assertTrue(uri.startswith("qemu+ssh://root@host/system?command="))
+        self.assertTrue(os.access(SSH_COMMAND, os.X_OK))
+        with open(SSH_COMMAND) as script:
+            self.assertIn("ConnectTimeout", script.read())
+
+
 if __name__ == "__main__":
     unittest.main()

vrtManager/test_live_connection.py

@@ -0,0 +1,249 @@
+"""Live checks of bounded connection opens against real libvirt.
+
+Runs only with TEST_LIBVIRT_HOST set (a TCP compute, see instances/livetest.py);
+TEST_LIBVIRT_SSH_HOST/_LOGIN add a reachable ssh compute. 192.0.2.1 is
+TEST-NET-1: SYN packets to it are dropped, as for a host that went away
+without a trace.
+
+TEST_LIBVIRT_SSH_ALIASES=1 adds the ssh tests that need these entries in the
+ssh configuration of the user running the tests (SSH_HOST is the ssh compute):
+
+    Host wvc-alias
+        HostName SSH_HOST
+    Host wvc-relay
+        HostName 127.0.0.1
+        Port 2222
+        HostKeyAlias SSH_HOST
+    Host wvc-silent
+        HostName 127.0.0.1
+        Port 2223
+
+TEST_LIBVIRT_SSH_JUMP=1 adds a ProxyJump test through
+    Host wvc-jump
+        HostName SSH_HOST
+        ProxyJump user@another-host
+"""
+
+import os
+import shutil
+import socket
+import subprocess
+import threading
+import time
+import unittest
+
+import libvirt
+from vrtManager.connection import CONN_SSH, CONN_TCP, CONNECT_TIMEOUT, connection_manager
+
+UNREACHABLE = "192.0.2.1"
+KEEPALIVE_DEATH = 45  # keepalive (5, 5) closes a silent connection after ~30 s
+
+
+def timed(*args):
+    started = time.monotonic()
+    try:
+        return connection_manager.get_connection(*args), time.monotonic() - started
+    except libvirt.libvirtError as err:
+        return err, time.monotonic() - started
+
+
+def ssh_processes(host):
+    """ssh client processes (exact name) whose arguments mention host."""
+    out = subprocess.run(["pgrep", "-a", "-x", "ssh"], capture_output=True, text=True).stdout
+    return [line for line in out.splitlines() if host in line]
+
+
+class Relay:
+    """TCP relay that can hold all traffic (frozen) and pass it again (thawed);
+    a frozen relay looks like a host that went silent."""
+
+    def __init__(self, target, port=0, frozen=False):
+        self.forwarding = threading.Event()
+        if not frozen:
+            self.forwarding.set()
+        self.server = socket.socket()
+        self.server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
+        self.server.bind(("127.0.0.1", port))
+        self.server.listen(5)
+        self.port = self.server.getsockname()[1]
+        self.target = target
+        self.closed = False
+        self.sockets = []
+        threading.Thread(target=self.accept, daemon=True).start()
+
+    def accept(self):
+        while True:
+            try:
+                client, _ = self.server.accept()
+            except OSError:
+                return
+            upstream = socket.create_connection(self.target)
+            self.sockets += [client, upstream]
+            for a, b in ((client, upstream), (upstream, client)):
+                threading.Thread(target=self.pump, args=(a, b), daemon=True).start()
+
+    def pump(self, src, dst):
+        try:
+            while not self.closed:
+                if not self.forwarding.wait(0.2):
+                    continue
+                src.settimeout(0.2)
+                try:
+                    data = src.recv(65536)
+                except socket.timeout:
+                    continue
+                if not data:
+                    break
+                dst.sendall(data)
+        except OSError:
+            pass
+
+    def freeze(self):
+        self.forwarding.clear()
+        time.sleep(0.5)  # let a recv that was already waiting run out
+
+    def thaw(self):
+        self.forwarding.set()
+
+    def close(self):
+        """Stop, and end the relayed connections, as a host that went away would."""
+        self.closed = True
+        self.forwarding.set()
+        try:
+            self.server.shutdown(socket.SHUT_RDWR)  # wakes the accept; close alone does not
+        except OSError:
+            pass
+        self.server.close()
+        for sock in self.sockets:
+            sock.close()
+
+
+@unittest.skipUnless(os.environ.get("TEST_LIBVIRT_HOST"), "Set TEST_LIBVIRT_HOST to run the live libvirt tests")
+class LiveBoundedOpenTestCase(unittest.TestCase):
+    def setUp(self):
+        self.host = os.environ["TEST_LIBVIRT_HOST"]
+        self.login = os.environ.get("TEST_LIBVIRT_LOGIN", "")
+        self.password = os.environ.get("TEST_LIBVIRT_PASSWORD", "")
+        self.ssh_host = os.environ.get("TEST_LIBVIRT_SSH_HOST")
+        self.ssh_login = os.environ.get("TEST_LIBVIRT_SSH_LOGIN", "root")
+
+    # Opening
+
+    def test_unreachable_tcp_host_fails_within_the_timeout_then_at_once(self):
+        args = (UNREACHABLE, "u", "p", CONN_TCP)
+        first, took = timed(*args)
+        self.assertIsInstance(first, libvirt.libvirtError)
+        self.assertLess(took, CONNECT_TIMEOUT + 1)
+        second, took = timed(*args)
+        self.assertIsInstance(second, libvirt.libvirtError)
+        self.assertLess(took, 0.5)
+
+    @unittest.skipUnless(shutil.which("ssh"), "no ssh client")
+    def test_unreachable_ssh_host_fails_within_the_timeout_and_leaves_no_ssh(self):
+        result, took = timed(UNREACHABLE, "root", "", CONN_SSH)
+        self.assertIsInstance(result, libvirt.libvirtError)
+        self.assertLess(took, CONNECT_TIMEOUT + 1)
+        time.sleep(12)  # the wrapper's ConnectTimeout is 10 s
+        self.assertEqual(ssh_processes(UNREACHABLE), [], "an ssh process outlived its connect timeout")
+
+    def test_a_daemon_that_never_answers_fails_within_the_timeout(self):
+        silent = socket.socket()
+        silent.bind(("127.0.0.1", 0))
+        silent.listen(5)  # accepts (kernel backlog), never answers
+        self.addCleanup(silent.close)
+        result, took = timed(f"127.0.0.1:{silent.getsockname()[1]}", "u", "p", CONN_TCP)
+        self.assertIsInstance(result, libvirt.libvirtError)
+        self.assertLess(took, CONNECT_TIMEOUT + 1)
+
+    def check_slow_open_completes_later(self, args, relay):
+        first, _ = timed(*args)
+        self.assertIsInstance(first, libvirt.libvirtError)
+        relay.thaw()
+        deadline = time.monotonic() + 15
+        while time.monotonic() < deadline:
+            conn, _ = timed(*args)
+            if not isinstance(conn, libvirt.libvirtError):
+                break
+            time.sleep(0.5)
+        self.assertNotIsInstance(conn, libvirt.libvirtError)
+        self.assertTrue(conn.getVersion())
+
+    def test_a_slow_tcp_open_that_completes_later_is_used(self):
+        relay = Relay((self.host, 16509), frozen=True)
+        self.addCleanup(relay.close)
+        self.check_slow_open_completes_later((f"127.0.0.1:{relay.port}", self.login, self.password, CONN_TCP), relay)
+
+    # A connection whose host goes silent
+
+    def check_silent_host_is_noticed_and_replaced(self, args, relay):
+        old = connection_manager.get_connection(*args)
+        relay.freeze()
+        outcome = []
+
+        def call():
+            try:
+                outcome.append(old.getVersion())
+            except libvirt.libvirtError as err:
+                outcome.append(err)
+
+        caller = threading.Thread(target=call, daemon=True)
+        caller.start()
+        caller.join(KEEPALIVE_DEATH)  # keepalive must end the call by then
+        if caller.is_alive():
+            relay.close()  # cut the connection so the stuck call returns
+            self.fail(f"a call on a silent connection still waits after {KEEPALIVE_DEATH} s")
+        self.assertIsInstance(outcome[0], libvirt.libvirtError)
+        relay.thaw()
+        new, _ = timed(*args)
+        self.assertNotIsInstance(new, libvirt.libvirtError)
+        self.assertIsNot(new, old)
+        self.assertTrue(new.getVersion())
+
+    def test_a_silent_tcp_host_is_noticed_and_reconnected(self):
+        relay = Relay((self.host, 16509))
+        self.addCleanup(relay.close)
+        self.check_silent_host_is_noticed_and_replaced((f"127.0.0.1:{relay.port}", self.login, self.password, CONN_TCP), relay)
+
+    # ssh
+
+    @unittest.skipUnless(os.environ.get("TEST_LIBVIRT_SSH_HOST"), "Set TEST_LIBVIRT_SSH_HOST (and _LOGIN) for an ssh compute")
+    def test_a_reachable_ssh_host_connects_through_the_wrapper(self):
+        conn, _ = timed(self.ssh_host, self.ssh_login, "", CONN_SSH)
+        self.assertNotIsInstance(conn, libvirt.libvirtError)
+        self.assertIn("command=", conn.getURI())
+        self.assertTrue(conn.getVersion())
+
+    @unittest.skipUnless(os.environ.get("TEST_LIBVIRT_SSH_ALIASES"), "Set TEST_LIBVIRT_SSH_ALIASES=1 (see the module docstring)")
+    def test_an_ssh_config_alias_is_used(self):
+        conn, _ = timed("wvc-alias", self.ssh_login, "", CONN_SSH)
+        self.assertNotIsInstance(conn, libvirt.libvirtError)
+        self.assertTrue(conn.getVersion())
+
+    @unittest.skipUnless(os.environ.get("TEST_LIBVIRT_SSH_ALIASES"), "Set TEST_LIBVIRT_SSH_ALIASES=1 (see the module docstring)")
+    def test_an_ssh_port_that_never_answers_fails_within_the_timeout(self):
+        silent = socket.socket()
+        silent.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
+        silent.bind(("127.0.0.1", 2223))
+        silent.listen(5)  # no ssh banner ever comes
+        self.addCleanup(silent.close)
+        result, took = timed("wvc-silent", self.ssh_login, "", CONN_SSH)
+        self.assertIsInstance(result, libvirt.libvirtError)
+        self.assertLess(took, CONNECT_TIMEOUT + 1)
+
+    @unittest.skipUnless(os.environ.get("TEST_LIBVIRT_SSH_ALIASES"), "Set TEST_LIBVIRT_SSH_ALIASES=1 (see the module docstring)")
+    def test_a_slow_ssh_open_that_completes_later_is_used(self):
+        relay = Relay((self.ssh_host, 22), port=2222, frozen=True)
+        self.addCleanup(relay.close)
+        self.check_slow_open_completes_later(("wvc-relay", self.ssh_login, "", CONN_SSH), relay)
+
+    @unittest.skipUnless(os.environ.get("TEST_LIBVIRT_SSH_ALIASES"), "Set TEST_LIBVIRT_SSH_ALIASES=1 (see the module docstring)")
+    def test_a_silent_ssh_host_is_noticed_and_reconnected(self):
+        relay = Relay((self.ssh_host, 22), port=2222)
+        self.addCleanup(relay.close)
+        self.check_silent_host_is_noticed_and_replaced(("wvc-relay", self.ssh_login, "", CONN_SSH), relay)
+
+    @unittest.skipUnless(os.environ.get("TEST_LIBVIRT_SSH_JUMP"), "Set TEST_LIBVIRT_SSH_JUMP=1 (see the module docstring)")
+    def test_ssh_through_a_jump_host(self):
+        conn, _ = timed("wvc-jump", self.ssh_login, "", CONN_SSH)
+        self.assertNotIsInstance(conn, libvirt.libvirtError)
+        self.assertTrue(conn.getVersion())