packages/cultmesh-py/src/cultmesh_py/server.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+import errno
 import logging
 import socket
 import threading
@@ -57,6 +58,7 @@ from cultnet_py.cultmesh_contracts import (
 )
 
 _LOGGER = logging.getLogger(__name__)
+_EPHEMERAL_PAIR_ATTEMPTS = 64
 
 
 class _ReplyUnqueueable(Exception):
@@ -168,26 +170,49 @@ class CultMeshLocalServer:
     def start(self) -> "CultMeshLocalServer":
         if self._socket is not None:
             return self
-        server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
-        server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
-        server.bind((self.host, self.port))
-        server.listen()
+        server, rudp_socket = self._bind_sockets()
         server.settimeout(0.2)
         self._socket = server
         self.port = int(server.getsockname()[1])
         self._stop.clear()
         self._thread = threading.Thread(target=self._accept_loop, daemon=True)
         self._thread.start()
-        if self.enable_rudp:
-            rudp_socket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
-            rudp_socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
-            rudp_socket.bind((self.host, self.port))
+        if rudp_socket is not None:
             rudp_socket.settimeout(0.02)
             self._rudp_socket = rudp_socket
             self._rudp_thread = threading.Thread(target=self._rudp_loop, daemon=True)
             self._rudp_thread.start()
         return self
 
+    def _bind_sockets(self) -> tuple[socket.socket, socket.socket | None]:
+        """The listening TCP socket and, when RUDP is on, the UDP socket on the same port number.
+
+        An explicit port is bound as asked and a failure is the caller's. Port 0 asks for any free port, and the
+        kernel picks the TCP one without knowing whether another process holds that number for UDP, so a
+        UDP collision retries with a fresh TCP port instead of failing the start. Both sockets are bound
+        before any thread starts, so a failed start leaves nothing running or open.
+        """
+        attempts = _EPHEMERAL_PAIR_ATTEMPTS if self.port == 0 and self.enable_rudp else 1
+        for attempt in range(attempts):
+            server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+            rudp_socket = None
+            try:
+                server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
+                server.bind((self.host, self.port))
+                server.listen()
+                if self.enable_rudp:
+                    rudp_socket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
+                    rudp_socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
+                    rudp_socket.bind((self.host, int(server.getsockname()[1])))
+                return server, rudp_socket
+            except OSError as error:
+                server.close()
+                if rudp_socket is not None:
+                    rudp_socket.close()
+                if error.errno != errno.EADDRINUSE or attempt == attempts - 1:
+                    raise
+        raise AssertionError("unreachable")
+
     def stop(self) -> None:
         self._stop.set()
         if self._rudp_socket is not None:

packages/cultmesh-py/tests/test_server_bind.py

@@ -0,0 +1,64 @@
+"""A server asked for any port finds a TCP+UDP pair; one asked for a port gets that port or an honest failure."""
+
+from __future__ import annotations
+
+import errno
+import socket
+import threading
+import unittest
+from unittest.mock import patch
+
+from cultmesh_py import create_node
+from cultmesh_py.server import CultMeshLocalServer
+
+
+def server(**kwargs) -> CultMeshLocalServer:
+    return CultMeshLocalServer(node=create_node(runtime_id="bind-test"), host="127.0.0.1", **kwargs)
+
+
+class ServerBindTests(unittest.TestCase):
+    def test_port_zero_retries_with_a_fresh_port_when_another_process_holds_the_udp_number(self) -> None:
+        real_bind = socket.socket.bind
+        refused: list[int] = []
+
+        def bind(sock: socket.socket, address) -> None:
+            # The first two UDP binds find the number taken, as a UDP listener of another process would.
+            if sock.type == socket.SOCK_DGRAM and len(refused) < 2:
+                refused.append(address[1])
+                raise OSError(errno.EADDRINUSE, "Address already in use")
+            real_bind(sock, address)
+
+        with patch.object(socket.socket, "bind", bind):
+            started = server().start()
+        try:
+            self.assertEqual(len(refused), 2)
+            self.assertNotIn(started.port, refused[:1], "the retry used a fresh TCP port")
+            probe = socket.create_connection(("127.0.0.1", started.port), timeout=2.0)
+            probe.close()
+        finally:
+            started.stop()
+
+    def test_an_explicit_port_whose_udp_number_is_taken_fails_and_leaves_nothing_running(self) -> None:
+        holder = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
+        holder.bind(("127.0.0.1", 0))
+        port = holder.getsockname()[1]
+        threads_before = threading.active_count()
+        try:
+            failing = server(port=port)
+            with self.assertRaises(OSError) as raised:
+                failing.start()
+            self.assertEqual(raised.exception.errno, errno.EADDRINUSE)
+            self.assertIsNone(failing._socket)
+            self.assertEqual(threading.active_count(), threads_before, "no accept thread was left running")
+            # The TCP number was released, not left listening.
+            free = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+            try:
+                free.bind(("127.0.0.1", port))
+            finally:
+                free.close()
+        finally:
+            holder.close()
+
+
+if __name__ == "__main__":
+    unittest.main()