test: SOCKS5 proxy: expect that connection may be reset when forwarding

diff --git a/test/functional/test_framework/netutil.py b/test/functional/test_framework/netutil.py
index 85209322..17043541 100644
--- a/test/functional/test_framework/netutil.py
+++ b/test/functional/test_framework/netutil.py
@@ -212,6 +212,37 @@ def format_addr_port(addr, port):
     else:
         return f"{addr}:{port}"
 
+def format_sock(sock, *, local):
+    '''
+    Format either local or remote side of a socket to a human readable string, e.g.
+    1.2.3.4:8333 or
+    [11:22::33]:8333 or
+    /path/to/socket or
+    @abstract-socket
+    '''
+    try:
+        if local:
+            name = sock.getsockname()
+        else:
+            name = sock.getpeername()
+    except Exception:
+        return "n/a"
+
+    if sock.family == socket.AF_INET:
+        return f"{name[0]}:{name[1]}"
+
+    if sock.family == socket.AF_INET6:
+        return f"[{name[0]}]:{name[1]}"
+
+    if sock.family == socket.AF_UNIX:
+        if isinstance(name, bytes):
+            name = name.decode(errors="backslashreplace")
+        if name.startswith("\0"):
+            return f"@{name[1:]}"
+        return name
+
+    return str(name)
+
 
 def set_ephemeral_port_range(sock):
     '''On FreeBSD, set socket to use the high ephemeral port range (49152-65535).
diff --git a/test/functional/test_framework/socks5.py b/test/functional/test_framework/socks5.py
index 930e0e67..a794dc08 100644
--- a/test/functional/test_framework/socks5.py
+++ b/test/functional/test_framework/socks5.py
@@ -12,6 +12,7 @@ import logging
 
 from .netutil import (
     format_addr_port,
+    format_sock,
     set_ephemeral_port_range,
 )
 
@@ -55,6 +56,12 @@ def forward_sockets(a, b, wakeup_socket, serv):
     Monitors wakeup_socket for a shutdown signal and checks serv.is_running()
     to exit gracefully when the server is stopping.
     """
+    # Prefix messages with e.g.:
+    # forward_sockets(a{remote=127.0.0.1:36935, local=127.0.0.1:9050} <-> b{local=127.0.0.1:33424, remote=127.0.0.1:8333})
+    log_prefix = ("forward_sockets("
+                  f"a{{remote={format_sock(a, local=False)}, local={format_sock(a, local=True)}}} <-> "
+                  f"b{{local={format_sock(b, local=True)}, remote={format_sock(b, local=False)}}}): ")
+
     # Mark as non-blocking so that we do not end up in a deadlock-like situation
     # where we block and wait on data from `a` while there is data ready to be
     # received on `b` and forwarded to `a`. And at the same time the application
@@ -63,24 +70,26 @@ def forward_sockets(a, b, wakeup_socket, serv):
     a.setblocking(False)
     b.setblocking(False)
     sockets = [a, b, wakeup_socket]
-    done = False
-    while not done:
+    while True:
         # Blocking select with timeout
         rlist, _, xlist = select.select(sockets, [], sockets, 2)
         if not serv.is_running():
-            logger.debug("forward_sockets: Exit due to shutdown")
+            logger.debug(f"{log_prefix}Exit due to shutdown")
             return
         if len(xlist) > 0:
-            raise IOError('Exceptional condition on socket')
+            raise IOError(f"{log_prefix}Exceptional condition on socket")
         for s in rlist:
-            data = s.recv(4096)
-            if data is None or len(data) == 0:
-                done = True
-                break
-            if s == a:
-                sendall(b, data)
-            elif s == b:
-                sendall(a, data)
+            try:
+                data = s.recv(4096)
+                if data is None or len(data) == 0:
+                    return
+                if s == a:
+                    sendall(b, data)
+                elif s == b:
+                    sendall(a, data)
+            except (BrokenPipeError, ConnectionResetError) as e:
+                logger.debug(f"{log_prefix}cannot send or receive data on socket {'a' if s == a else 'b'}: {str(e)}")
+                return
 
 # Implementation classes
 class Socks5Configuration():