fix: jsocket reconnect lifecycle 2.0.2

jsocket/_version.py

@@ -1,3 +1,3 @@
 """Package version."""
 
-__version__ = "2.0.1"
+__version__ = "2.0.2"

jsocket/jsocket_base.py

@@ -71,6 +71,7 @@ def __init__(
         self.max_message_size = max_message_size
         self._last_client_addr = None
         self._is_server = False
+        self._is_listening = False
         if create_socket:
             self.socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
             self.conn = self.socket
@@ -93,7 +94,15 @@ def _send(self, msg):
         """Send all bytes in `msg` to the peer."""
         sent = 0
         while sent < len(msg):
-            sent += self.conn.send(msg[sent:])
+            try:
+                chunk = self.conn.send(msg[sent:])
+            except OSError as e:
+                self._close_connection()
+                raise RuntimeError("socket connection broken") from e
+            if chunk == 0:
+                self._close_connection()
+                raise RuntimeError("socket connection broken")
+            sent += chunk
 
     def _read(self, size, allow_timeout=False):
         """Read exactly `size` bytes from the peer or raise on disconnect."""
@@ -168,6 +177,7 @@ def _close_socket(self):
                     pass
         except OSError:
             pass
+        self._is_listening = False
 
     def _close_connection(self):
         """Best-effort shutdown and close of the connection socket."""
@@ -301,9 +311,13 @@ def _close_connection(self):
     def _bind(self):
         self.socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
         self.socket.bind((self.address, self.port))
+        self._is_listening = False
 
     def _listen(self):
+        if self._is_listening:
+            return
         self.socket.listen(5)
+        self._is_listening = True
 
     def _accept(self):
         return self.socket.accept()
@@ -341,17 +355,30 @@ def __init__(self, address='127.0.0.1', port=5489, timeout=2.0, recv_timeout=Non
 
     def connect(self):
         """Attempt to connect to the server up to 10 times with backoff."""
+        def _recreate_socket():
+            self._close_socket()
+            self.socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+            self.socket.settimeout(self._recv_timeout)
+            self.conn = self.socket
+
         for attempt in range(1, 11):
+            sock = getattr(self, "socket", None)
+            needs_fresh_socket = sock is None
+            if not needs_fresh_socket:
+                try:
+                    needs_fresh_socket = sock.fileno() == -1
+                except OSError:
+                    needs_fresh_socket = True
+            if needs_fresh_socket:
+                _recreate_socket()
+                logger.debug("created fresh socket before connect attempt %d to %s:%s", attempt, self.address, self.port)
             try:
                 logger.debug("connect attempt %d to %s:%s", attempt, self.address, self.port)
                 self.socket.connect((self.address, self.port))
             except socket.error as msg:
                 logger.error("SockThread Error: %s", msg)
                 # Recreate the socket to avoid retrying on a potentially bad fd.
-                self._close_socket()
-                self.socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
-                self.socket.settimeout(self._recv_timeout)
-                self.conn = self.socket
+                _recreate_socket()
                 logger.debug("recreated socket for retry %d to %s:%s", attempt, self.address, self.port)
                 time.sleep(3)
                 continue

jsocket/tserver.py

@@ -76,6 +76,9 @@ def _init_wakeup(self):
         except (AttributeError, OSError):
             pass
         # Fallback: create a loopback TCP pair
+        listener = None
+        writer = None
+        reader = None
         try:
             listener = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
             listener.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
@@ -84,12 +87,24 @@ def _init_wakeup(self):
             writer = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
             writer.connect(listener.getsockname())
             reader, _ = listener.accept()
-            listener.close()
             reader.setblocking(False)
             writer.setblocking(False)
             self._wakeup_r, self._wakeup_w = reader, writer
         except OSError:
+            for sock in (reader, writer):
+                if sock is None:
+                    continue
+                try:
+                    sock.close()
+                except OSError:
+                    pass
             self._wakeup_r, self._wakeup_w = None, None
+        finally:
+            if listener is not None:
+                try:
+                    listener.close()
+                except OSError:
+                    pass
 
     def _signal_wakeup(self):
         """Wake the accept loop if it is blocked."""
@@ -398,7 +413,6 @@ def run(self):
         if not self._is_alive:
             self._is_alive = True
         while self._is_alive:
-            tmp = self._thread_type(**self._thread_args)
             self._purge_threads()
             while not self.connected and self._is_alive:
                 if not self._wait_for_accept():
@@ -409,7 +423,10 @@ def run(self):
                     logger.debug("factory accept timeout on %s:%s: %s", self.address, self.port, e)
                     continue
                 except Exception as e:  # pylint: disable=broad-exception-caught
-                    logger.exception("accept error: %s", e)
+                    if self._is_alive:
+                        logger.exception("factory accept error on %s:%s: %s", self.address, self.port, e)
+                    else:
+                        logger.debug("factory stopping; accept loop exiting (%s:%s)", self.address, self.port)
                     continue
                 else:
                     # Hand off the accepted connection to the worker
@@ -427,10 +444,42 @@ def run(self):
                         except OSError:
                             pass
                         break
-                    tmp.swap_socket(accepted_conn)
-                    tmp.start()
+                    try:
+                        tmp = self._thread_type(**self._thread_args)
+                        tmp.swap_socket(accepted_conn)
+                        tmp.start()
+                    except Exception as e:  # pylint: disable=broad-exception-caught
+                        # Worker construction/hand-off failed; ensure the accepted connection is closed
+                        try:
+                            accepted_conn.shutdown(socket.SHUT_RDWR)
+                        except OSError:
+                            pass
+                        try:
+                            accepted_conn.close()
+                        except OSError:
+                            pass
+                        if self._is_alive:
+                            logger.exception(
+                                "factory worker handoff error on %s:%s: %s",
+                                self.address,
+                                self.port,
+                                e,
+                            )
+                        else:
+                            logger.debug(
+                                "factory stopping; worker handoff aborted (%s:%s)",
+                                self.address,
+                                self.port,
+                            )
+                        # Continue accepting new connections
+                        continue
                     with self._threads_lock:
                         self._threads.append(tmp)
+                    try:
+                        addr = accepted_conn.getpeername()
+                    except OSError:
+                        addr = None
+                    logger.debug("factory spawned worker %s for %s", tmp.name, _format_client_id(addr))
                     break
 
         self._wait_to_exit()

tests/test_additional_coverage.py

@@ -29,6 +29,67 @@ def fail_connect(self, addr):  # pylint: disable=unused-argument
     assert client.connect() is False
 
 
+def test_client_connect_recreates_closed_socket_without_backoff(monkeypatch):
+    """connect() should refresh a closed fd before dialing, avoiding backoff sleep."""
+
+    events = []
+
+    class ClosedSocket:
+        def fileno(self):
+            return -1
+
+        def settimeout(self, timeout):
+            events.append(("closed_settimeout", timeout))
+
+        def connect(self, addr):  # pragma: no cover - should never be called
+            events.append(("closed_connect", addr))
+            raise AssertionError("connect() attempted on closed socket")
+
+        def shutdown(self, how):  # pylint: disable=unused-argument
+            return None
+
+        def close(self):
+            return None
+
+    class FreshSocket:
+        def fileno(self):
+            return 42
+
+        def settimeout(self, timeout):
+            events.append(("fresh_settimeout", timeout))
+
+        def connect(self, addr):
+            events.append(("fresh_connect", addr))
+            return None
+
+        def shutdown(self, how):  # pylint: disable=unused-argument
+            return None
+
+        def close(self):
+            return None
+
+    def fake_socket(*args, **kwargs):  # pylint: disable=unused-argument
+        return FreshSocket()
+
+    def no_sleep(*args, **kwargs):  # pylint: disable=unused-argument
+        raise AssertionError("unexpected connect backoff sleep")
+
+    monkeypatch.setattr(jsocket.jsocket_base.socket, "socket", fake_socket, raising=True)
+    monkeypatch.setattr(jsocket.jsocket_base.time, "sleep", no_sleep, raising=True)
+
+    client = jsocket.jsocket_base.JsonClient.__new__(jsocket.jsocket_base.JsonClient)
+    client.socket = ClosedSocket()
+    client.conn = client.socket
+    client._recv_timeout = 0.25
+    client._address = "127.0.0.1"
+    client._port = 5489
+    client._is_listening = False
+
+    assert client.connect() is True
+    assert ("fresh_connect", ("127.0.0.1", 5489)) in events
+    assert not [e for e in events if e[0] == "closed_connect"]
+
+
 def test_close_idempotent_and_connected_guard():
     """close() is safe to call multiple times; connected guard tolerates None."""
     # Client close idempotence