Overall connect deadline for boards that flood output

cli/src/mpftp/sidecar.py

@@ -203,6 +203,76 @@ def _bound_write_timeout(transport: Any, secs: float = WRITE_TIMEOUT_SECS) -> No
             pass
 
 
+#: Longest one connect attempt may spend taking control of a serial board, in
+#: seconds. mpremote's timeouts are between bytes, so a board that prints
+#: forever never trips them. ``MPFTP_CONNECT_DEADLINE`` overrides it.
+CONNECT_DEADLINE_SECS = 45.0
+
+
+def connect_deadline_secs() -> float:
+    try:
+        return float(os.environ.get("MPFTP_CONNECT_DEADLINE") or CONNECT_DEADLINE_SECS)
+    except ValueError:
+        return CONNECT_DEADLINE_SECS
+
+
+class ConnectDeadlineError(RuntimeError):
+    """The connect deadline passed while the board was still sending bytes."""
+
+
+class _DeadlineSerial:
+    """A serial port that refuses everything once a deadline has passed.
+
+    Wraps the port only while connect takes control, so a flood of output that
+    keeps mpremote's per-byte timeout from ever firing still ends the attempt.
+    After the deadline even a line-control write (the reset path) is refused.
+    """
+
+    def __init__(self, inner: Any, deadline: float) -> None:
+        object.__setattr__(self, "_inner", inner)
+        object.__setattr__(self, "_deadline", deadline)
+
+    def _check(self) -> None:
+        import time
+
+        if time.monotonic() >= self._deadline:
+            raise ConnectDeadlineError("connect deadline passed")
+
+    def __getattr__(self, name: str) -> Any:
+        self._check()
+        return getattr(self._inner, name)
+
+    def __setattr__(self, name: str, value: Any) -> None:
+        self._check()
+        setattr(self._inner, name, value)
+
+
+def _serial_holder(transport: Any) -> tuple[Any, str]:
+    """Where the raw pyserial port sits, so it can be wrapped in place.
+
+    Our transport reads through a ReadAheadSerial that keeps buffered bytes, so
+    the port inside it is wrapped, not the transport's ``serial`` (setting that
+    would build a new, empty ReadAheadSerial). Any other transport holds the
+    port in ``serial``.
+    """
+    ser = getattr(transport, "serial", None)
+    if ser is None:
+        return None, ""
+    if "_port" in getattr(ser, "__dict__", {}):
+        return ser, "_port"
+    return transport, "serial"
+
+
+def connect_deadline_message(device: str, secs: float) -> str:
+    return (
+        f"{device} kept sending output for {secs:g} s and never answered the REPL, "
+        f"so mpftp gave up. A program that prints continuously (a timer callback "
+        f"that never stops, say) keeps the board from listening. A board with only "
+        f"its native USB for serial can't be reset from the host: press its RESET "
+        f"button, then try again."
+    )
+
+
 def is_eof_timeout_error(exc: BaseException) -> bool:
     """True when mpremote follow-exec timed out waiting for raw-REPL EOF."""
     msg = str(exc).lower()
@@ -1001,9 +1071,28 @@ def connect(
                     raise RuntimeError(self._friendly_port_open_error(device, e)) from e
                 # Opening the COM port succeeds in UF2/bootloader mode too — require a
                 # MicroPython raw-REPL handshake before we report connected.
+                bounded = None
                 try:
-                    rtc = self._probe_micropython(self.transport)
+                    secs = connect_deadline_secs()
+                    holder, attr = _serial_holder(self.transport)
+                    if holder is not None and not network:
+                        raw_serial = object.__getattribute__(holder, attr)
+                        bounded = _DeadlineSerial(raw_serial, time.monotonic() + secs)
+                        object.__setattr__(holder, attr, bounded)
+                    try:
+                        rtc = self._probe_micropython(self.transport)
+                    finally:
+                        if bounded is not None:
+                            object.__setattr__(holder, attr, raw_serial)
                 except Exception as e:
+                    if bounded is not None and time.monotonic() >= bounded._deadline:
+                        try:
+                            self.transport.close()
+                        except Exception:
+                            pass
+                        self.transport = None
+                        self.device = None
+                        raise RuntimeError(connect_deadline_message(device, secs)) from e
                     trace("attempt %d: probe failed: %s" % (attempt, str(e).splitlines()[0][:80]))
                     try:
                         self.transport.close()

cli/tests/test_connect_deadline.py

@@ -0,0 +1,109 @@
+"""A board that floods output forever must not hold connect indefinitely (mpftp#193).
+
+mpremote's timeouts run between bytes, so continuous output defeats them.
+No board required: a fake transport streams forever.
+"""
+
+from __future__ import annotations
+
+import os
+import time
+import unittest
+from unittest import mock
+
+
+class FloodingSerial:
+    """Always has a byte to give, like a Timer printing every few ms."""
+
+    def __init__(self):
+        self.writes = []
+        self.dtr = False
+
+    def inWaiting(self):
+        return 64
+
+    def read(self, n=1):
+        time.sleep(0.001)
+        return b"tick tick tick\r\n"[:n] if n < 16 else b"tick tick tick\r\n"
+
+    def write(self, data):
+        self.writes.append(data)
+        return len(data)
+
+    def close(self):
+        pass
+
+
+class FloodingTransport:
+    def __init__(self):
+        self.serial = FloodingSerial()
+        self.in_raw_repl = False
+        self.closed = False
+
+    def enter_raw_repl(self, soft_reset=True, timeout_overall=10.0):
+        # mpremote's read_until: restarts its per-byte timer on every byte.
+        while True:
+            self.serial.read(1)
+
+    def close(self):
+        self.closed = True
+
+
+class ConnectDeadlineTests(unittest.TestCase):
+    def test_endless_flood_ends_with_a_reset_button_message(self):
+        from mpftp import sidecar
+
+        s = sidecar.Session()
+        t = FloodingTransport()
+        started = time.monotonic()
+        with mock.patch.dict(os.environ, {"MPFTP_CONNECT_DEADLINE": "1.5"}), \
+                mock.patch.object(s, "_open_transport", return_value=t), \
+                mock.patch.object(s, "_serial_peek", return_value=b""), \
+                self.assertRaises(RuntimeError) as cm:
+            s.connect("COM99", attempts=3)
+        took = time.monotonic() - started
+        msg = str(cm.exception)
+        self.assertIn("RESET button", msg)
+        self.assertIn("kept sending output", msg)
+        self.assertLess(took, 6)
+        self.assertTrue(t.closed)
+        self.assertIsNone(s.transport)
+
+    def test_flood_inside_our_read_ahead_transport_is_cut_off(self):
+        # The real transport reads through ReadAheadSerial.read_until, which
+        # checks timeout_overall only while idle, so a flood stays inside it.
+        from mpftp import rawpaste, sidecar
+
+        s = sidecar.Session()
+        cls = rawpaste._transport_class()
+        t = cls.__new__(cls)
+        t.in_raw_repl = False
+        t.use_raw_paste = True
+        t.device_name = "COM99"
+        t.mounted = False
+        t.is_pty = False
+        t.paced = None
+        t.serial = FloodingSerial()
+        started = time.monotonic()
+        with mock.patch.dict(os.environ, {"MPFTP_CONNECT_DEADLINE": "1.5"}), \
+                mock.patch.object(s, "_open_transport", return_value=t), \
+                mock.patch.object(s, "_serial_peek", return_value=b""), \
+                self.assertRaises(RuntimeError) as cm:
+            s.connect("COM99", attempts=3)
+        self.assertIn("RESET button", str(cm.exception))
+        self.assertLess(time.monotonic() - started, 8)
+
+    def test_the_port_is_unwrapped_after_a_good_connect(self):
+        from mpftp import sidecar
+
+        s = sidecar.Session()
+        t = FloodingTransport()
+        raw = t.serial
+        with mock.patch.object(s, "_open_transport", return_value=t), \
+                mock.patch.object(s, "_probe_micropython", return_value=None):
+            s.connect("COM99", attempts=1)
+        self.assertIs(t.serial, raw)
+
+
+if __name__ == "__main__":
+    unittest.main()