Merge pull request #958 from meshtastic/tcp-graceful-close

Half-close TCP connections so the last write is not lost

meshtastic/tcp_interface.py

@@ -11,6 +11,11 @@ from typing import Optional
 from meshtastic.stream_interface import StreamInterface
 
 DEFAULT_TCP_PORT = 4403
+
+# How long close() gives the device to consume what we last wrote before forcing
+# the connection down. See close() for why this exists.
+GRACEFUL_CLOSE_TIMEOUT = 0.25
+
 logger = logging.getLogger(__name__)
 
 
@@ -80,6 +85,18 @@ class TCPInterface(StreamInterface):
         server_address = (self.hostname, self.portNumber)
         self.socket = socket.create_connection(server_address)
 
+    def _wait_for_reader_exit(self, timeout: float) -> None:
+        """Wait briefly for the reader thread to drain and exit after a half-close.
+
+        Returns as soon as it exits, or after timeout: the device is not obliged
+        to close just because we did.
+        """
+        rx = getattr(self, "_rxThread", None)
+        if rx is None or rx is threading.current_thread():
+            return
+        with contextlib.suppress(Exception):
+            rx.join(timeout)
+
     def close(self) -> None:
         """Close a connection to the device."""
         logger.debug("Closing TCP stream")
@@ -87,6 +104,20 @@ class TCPInterface(StreamInterface):
         # Therefore force a shutdown first to unblock reader thread reads.
         self._wantExit = True
         if self.socket is not None:
+            # Half-close first. shutdown(SHUT_WR) sends FIN, which tells the
+            # device we are done writing and lets it consume what we last wrote
+            # before the connection goes away -- typically the admin message a
+            # one-shot command such as `--set` just sent.
+            #
+            # Going straight to shutdown(SHUT_RDWR) + close() while either side
+            # still has unread data makes the stack send RST instead. Winsock
+            # then discards data the peer had already received but not yet read,
+            # so the write is lost; Linux delivers it before reporting
+            # ECONNRESET, which is why this only bites on Windows.
+            with contextlib.suppress(Exception):
+                self.socket.shutdown(socket.SHUT_WR)
+            self._wait_for_reader_exit(GRACEFUL_CLOSE_TIMEOUT)
+
             with contextlib.suppress(
                 Exception
             ):  # Ignore errors in shutdown, because we might have a race with the server

meshtastic/tests/test_tcp_interface.py

@@ -1,6 +1,7 @@
 """Meshtastic unit tests for tcp_interface.py"""
 
 import re
+import socket
 from unittest.mock import MagicMock, patch
 
 import pytest
@@ -78,6 +79,56 @@ def test_TCPInterface_close_shutdowns_socket_before_super_close():
     assert iface.socket is None
 
 
+@pytest.mark.unit
+def test_TCPInterface_close_half_closes_before_shutdown():
+    """Close should send FIN and let the device drain before forcing the link down.
+
+    Going straight to shutdown(SHUT_RDWR)/close() with data still unread makes the
+    stack send RST, and Winsock then discards data the device had received but not
+    yet read, silently losing writes such as `--set`.
+    """
+    iface = TCPInterface(hostname="localhost", noProto=True, connectNow=False)
+    sock = MagicMock()
+    iface.socket = sock
+    call_order = []
+
+    with patch.object(TCPInterface, "_socket_shutdown", autospec=True) as mock_shutdown:
+        with patch.object(
+            TCPInterface, "_wait_for_reader_exit", autospec=True
+        ) as mock_wait:
+            with patch(
+                "meshtastic.stream_interface.StreamInterface.close", autospec=True
+            ) as mock_super_close:
+                sock.shutdown.side_effect = lambda how: call_order.append(f"shutdown_{how}")
+                mock_wait.side_effect = lambda _self, _t: call_order.append("wait")
+                mock_shutdown.side_effect = lambda _self: call_order.append("full_shutdown")
+                mock_super_close.side_effect = lambda _self: call_order.append("super_close")
+
+                iface.close()
+
+    assert call_order == [
+        f"shutdown_{socket.SHUT_WR}",
+        "wait",
+        "full_shutdown",
+        "super_close",
+    ]
+
+
+@pytest.mark.unit
+def test_TCPInterface_close_survives_half_close_failure():
+    """A peer that already vanished must not break close()."""
+    iface = TCPInterface(hostname="localhost", noProto=True, connectNow=False)
+    sock = MagicMock()
+    sock.shutdown.side_effect = OSError("already gone")
+    iface.socket = sock
+
+    with patch("meshtastic.stream_interface.StreamInterface.close", autospec=True):
+        iface.close()  # must not raise
+
+    sock.close.assert_called_once()
+    assert iface.socket is None
+
+
 @pytest.mark.unit
 def test_TCPInterface_reconnect():
     """Test that _reconnect correctly reconnects"""