bledev: fast reconnects, and REPL output that arrives intact

docs/bledev-internals.md

@@ -181,6 +181,27 @@ no difference to nus throughput.
 After a disconnect bleak sets `client.services` to `None`, so every client call
 checks the connection first and raises `DisconnectedError`.
 
+## Reconnecting fast
+
+The REPL gate once failed one run in five: bleak couldn't subscribe on a
+connection made 1.5 s after the last one closed. A loop of rapid reconnects
+from the laptop (`tests/bledev_board/reconnect_loop.py`: log in, run a line,
+close, wait 1.5 s) caught it at about one cycle in fifty, always the same
+way. bleak's connect returned in 70-440 ms where a real one takes 1.7 s or
+more, Windows reported the device connected and the GATT session active, the
+board never saw a connection (its IRQ log shows it advertising throughout),
+and the first GATT operation hung until Windows reported a drop about nine
+seconds later.
+
+It went away with the REPL's output fixes (below): against `bledev-host`'s
+`repl.py` the loop failed 7 times in 465 cycles; against the fixed one, 0 in
+270, 150 of them with the retry switched off. We haven't found how the
+board's old timer handling produced a connection Windows believed in and the
+board never saw, so `bledev.connect_and_set_up()` stays as the defence: it
+bounds the setup at 4 s and tries again, up to three times. `nus.connect()`
+and `midi.connect()` use it, and the contract checks both the retry and the
+giving up.
+
 ## How the REPL gets there
 
 The REPL has to work when nothing runs asyncio, at the `>>>` prompt, so the
@@ -190,9 +211,26 @@ with `gap_advertise`, and does everything else from an IRQ handler it adds to
 aioble's dispatcher (or sets itself, with no aioble). `os.dupterm` reads a
 `_Stream` whose `readinto` returns `None` when empty, because 0 means end of
 stream and detaches it. Output goes out as notifications of `mtu - 3` bytes,
-and what the controller can't take yet is retried from a one-shot soft timer.
-A program printing faster than the link carries waits in `write()` up to two
-seconds, then drops the excess, because `print()` can't fail.
+and what the controller can't take yet is retried every 5 ms by one periodic
+`machine.Timer`. A program printing faster than the link carries waits in
+`write()` up to two seconds, then drops the excess, because `print()` can't
+fail.
+
+Two ways that retry went wrong, both found by printing a 16,890-character
+list over the link (the REPL gate now does this):
+
+- **A doubled chunk and a lost one.** The timer's callback runs between the
+  main program's bytecodes, so it could land inside a `flush()` that
+  `print()` had started: both sent the same chunk, then both trimmed it, and
+  the next chunk never went out. Five prints out of five came back corrupt.
+  `flush()` now returns at once if it's already running.
+- **A panic.** The retry used to be a one-shot timer re-armed on every write.
+  On the esp32, `Timer.init()` on a virtual timer whose alarm is being
+  dispatched clears its handler first, so `esp_timer`'s task on the other
+  core calls a NULL function (`InstrFetchProhibited` in
+  `timer_process_alarm`). It took the board down twice in about 45 long
+  prints. The timer is now started once and never re-armed; 60 long prints
+  afterwards, no panic and every byte intact.
 
 Ctrl-C works because the IRQ calls `os.dupterm_notify()` whenever a write
 contains 0x03; dupterm then reads it and raises `KeyboardInterrupt`, even in

lib/bledev/__init__.py

@@ -870,3 +870,38 @@ def _want_response(self, response):
 def is_adapter(obj):
     """True for a bledev :class:`BLE`; False for a raw ``bluetooth.BLE`` or anything else."""
     return isinstance(obj, BLE)
+
+
+async def connect_and_set_up(
+    ble, setup, *, name=None, service=None, device=None, timeout_ms=10000, attempts=3, setup_timeout_ms=4000, **options
+):
+    """Find a device, connect, and run ``await setup(connection)``; returns
+    ``(connection, what setup returned)``.
+
+    If the link drops or stalls before ``setup`` finishes, this disconnects
+    and tries again, up to ``attempts`` times. Profiles use it (``nus``,
+    ``midi``), because a link can come up dead: about one reconnect in fifty
+    from Windows, the OS reports a connection the peripheral never saw, and
+    the first GATT operation on it hangs until Windows gives up nine seconds
+    later. ``setup_timeout_ms`` bounds that wait. ``options`` go to
+    ``device.connect()``.
+    """
+    error = None
+    for _ in range(attempts):
+        target = device
+        if target is None:
+            target = await ble.find(name=name, service=service, timeout_ms=timeout_ms)
+        connection = await target.connect(timeout_ms=timeout_ms, **options)
+        try:
+            return connection, await wait_ms(setup(connection), setup_timeout_ms, "setting up the connection")
+        except (DisconnectedError, BLETimeoutError) as e:
+            error = e
+        except BaseException:
+            if connection.is_connected():
+                await connection.disconnect()
+            raise
+        try:
+            await connection.disconnect()
+        except BLEError:
+            pass
+    raise error

lib/bledev/nus.py

@@ -38,6 +38,7 @@
     Service,
     UnsupportedError,
     UUID,
+    connect_and_set_up,
     is_adapter,
     sleep_ms,
 )
@@ -351,24 +352,23 @@ async def connect(ble, name=None, *, device=None, timeout_ms=10000, mtu=MTU, **c
     """
     ble = _adapter(ble)
     _set_mtu(ble, mtu)
-    if device is None:
-        device = await ble.find(name=name, service=SERVICE, timeout_ms=timeout_ms)
-    connection = await device.connect(timeout_ms=timeout_ms, **connect_options)
-    try:
+
+    async def setup(connection):
         try:
             await connection.exchange_mtu(mtu)
         except (UnsupportedError, BLETimeoutError):
             pass
         service = await connection.service(SERVICE)
         if service is None:
-            raise BLEError("{!r} has no Nordic UART service".format(device))
+            raise BLEError("{!r} has no Nordic UART service".format(connection.device))
         rx = await service.characteristic(RX)
         tx = await service.characteristic(TX)
         if rx is None or tx is None:
-            raise BLEError("{!r}: Nordic UART service is missing RX or TX".format(device))
+            raise BLEError("{!r}: Nordic UART service is missing RX or TX".format(connection.device))
         await tx.subscribe(notify=True)
-    except BaseException:
-        if connection.is_connected():
-            await connection.disconnect()
-        raise
+        return rx, tx
+
+    connection, (rx, tx) = await connect_and_set_up(
+        ble, setup, name=name, service=SERVICE, device=device, timeout_ms=timeout_ms, **connect_options
+    )
     return _ClientLink(connection, rx, tx)

lib/bledev/repl.py

@@ -34,6 +34,9 @@
 from . import nus
 
 PROMPT = b"Password: "
+
+# How often the server retries output the controller couldn't take yet.
+_TICK_MS = 5
 BANNER = b"\r\nbledev REPL connected\r\n>>> "
 DENIED = b"\r\nAccess denied\r\n"
 
@@ -190,10 +193,12 @@ def __init__(self, password, name, interval_us, dupterm_index):
         self.rx = bytearray()
         self.rx_pos = 0
         self.tx = bytearray()
+        self.flushing = False
         self.dropped_out = 0
         self.prev_term = None
         self.stream = _Stream(self)
         self.timer = None
+        self.hang_up_at = None
         self.early_mtu = None
 
     # -- setup
@@ -236,8 +241,12 @@ def start(self):
         # Appending: several writes between two IRQs arrive together.
         ble.gatts_set_buffer(self.rx_handle, 1024, True)
         self._forget_other_services()
+        # One periodic timer for the life of the server, never re-armed: on
+        # the esp32, re-initialising a virtual Timer while its alarm is being
+        # dispatched calls a NULL handler and panics the board (seen as a
+        # Guru Meditation in esp_timer's task during long prints).
         self.timer = machine.Timer(-1)
-        self.one_shot = machine.Timer.ONE_SHOT
+        self.timer.init(mode=machine.Timer.PERIODIC, period=_TICK_MS, callback=self._tick)
         self.adv = pack_advertisement(self.name, [nus.SERVICE])
         self.running = True
         self.advertise()
@@ -292,6 +301,7 @@ def on_connect(self, conn):
             self.mtu = self.early_mtu[1]
         self.early_mtu = None
         self.login = _Gate(self.password)
+        self.hang_up_at = None
         self.authed = False
         self.rx = bytearray()
         self.rx_pos = 0
@@ -305,6 +315,7 @@ def on_disconnect(self, conn):
             return
         self._detach()
         self.conn = None
+        self.hang_up_at = None
         self.login = None
         self.authed = False
         self.rx = bytearray()
@@ -321,7 +332,9 @@ def on_write(self, conn, data):
                 self.notify_now(reply)
                 self.login.refused = True
                 # Let "Access denied" reach the client, then hang up.
-                self.timer.init(mode=self.one_shot, period=300, callback=self._hang_up)
+                import time
+
+                self.hang_up_at = time.ticks_add(time.ticks_ms(), 300)
                 return
             if verdict is None:
                 if reply:
@@ -373,7 +386,18 @@ def feed_repl(self, data):
 
     # -- sending
 
-    def _hang_up(self, _timer=None):
+    def _tick(self, _timer=None):
+        # Every _TICK_MS while the server runs, as a scheduled callback.
+        if self.hang_up_at is not None:
+            import time
+
+            if time.ticks_diff(time.ticks_ms(), self.hang_up_at) >= 0:
+                self.hang_up_at = None
+                self._hang_up()
+        if self.tx and self.conn is not None:
+            self.flush()
+
+    def _hang_up(self):
         if self.conn is not None and not self.authed:
             try:
                 self.ble.gap_disconnect(self.conn)
@@ -409,23 +433,26 @@ def send(self, data):
                     break
                 time.sleep_ms(2)
                 self.flush()
-        if self.tx:
-            self.timer.init(mode=self.one_shot, period=5, callback=self._flush_later)
-
-    def _flush_later(self, _timer=None):
-        self.flush()
-        if self.tx and self.conn is not None:
-            self.timer.init(mode=self.one_shot, period=5, callback=self._flush_later)
 
     def flush(self):
-        size = self.mtu - 3
-        while self.tx and self.conn is not None:
-            chunk = self.tx[:size]
-            try:
-                self.ble.gatts_notify(self.conn, self.tx_handle, chunk)
-            except OSError:
-                return  # the controller's buffers are full; the timer retries
-            self.tx = self.tx[len(chunk) :]
+        # The timer's callback runs between the main program's
+        # bytecodes, so it can land inside a flush() that print() started.
+        # Two flushes interleaved sent one chunk twice and dropped the next,
+        # so the second one leaves the work to the first.
+        if self.flushing:
+            return
+        self.flushing = True
+        try:
+            size = self.mtu - 3
+            while self.tx and self.conn is not None:
+                chunk = self.tx[:size]
+                try:
+                    self.ble.gatts_notify(self.conn, self.tx_handle, chunk)
+                except OSError:
+                    return  # the controller's buffers are full; _tick retries
+                self.tx = self.tx[len(chunk) :]
+        finally:
+            self.flushing = False
 
 
 class _Gate(Login):

tests/bledev_board/reconnect_loop.py

@@ -0,0 +1,72 @@
+# SPDX-FileCopyrightText: 2026 Brad Barnett
+#
+# SPDX-License-Identifier: MIT
+"""Rapid reconnects from a laptop against the REPL gate's board.
+
+Needs ``repl_server.py`` serving (from ``/main.py``). Each cycle logs in
+through ``bledev.repl.connect()``, runs one line, closes, and waits ``GAP``
+seconds before the next, which is how the REPL gate once failed: bleak
+couldn't subscribe on a connection made 1.5 s after the last one closed.
+Prints each cycle, how many connection attempts it took, and a total::
+
+    PYTHONPATH="$(wslpath -w lib)" python.exe "$(wslpath -w tests/bledev_board/reconnect_loop.py)" 100 1.5
+
+``--no-retry`` sets ``attempts=1``, the old behaviour, to show the failure.
+"""
+import asyncio
+import sys
+import time
+
+import bledev
+import bledev.bleak
+import bledev.nus as nus
+import bledev.repl as repl
+
+args = [a for a in sys.argv[1:] if not a.startswith("--")]
+N = int(args[0]) if args else 20
+GAP = float(args[1]) if len(args) > 1 else 1.5
+RETRY = "--no-retry" not in sys.argv
+
+attempts = []
+_helper = nus.connect_and_set_up
+
+
+async def counting(ble, setup, **kwargs):
+    async def counted(connection):
+        attempts[-1] += 1
+        return await setup(connection)
+
+    if not RETRY:
+        kwargs["attempts"] = 1
+    return await _helper(ble, counted, **kwargs)
+
+
+nus.connect_and_set_up = counting
+
+
+async def main():
+    ble = bledev.bleak.BleakBLE()
+    fails = 0
+    retried = 0
+    for i in range(N):
+        attempts.append(0)
+        t0 = time.monotonic()
+        try:
+            link = await repl.connect(ble, "bledev-gate-pw", name="bledev-repl", timeout_ms=15000)
+            out = await repl.run(link, "%d * 3" % i)
+            ok = out.strip() == str(i * 3)
+            await link.close()
+            status = "OK" if ok else "BAD OUTPUT"
+            fails += 0 if ok else 1
+        except Exception as e:
+            fails += 1
+            status = "FAIL {} {}".format(type(e).__name__, e)
+        if attempts[-1] > 1:
+            retried += 1
+        print("cycle", i, status, "attempts", attempts[-1], "%.1fs" % (time.monotonic() - t0), flush=True)
+        await asyncio.sleep(GAP)
+    print("RESULT {}/{} failed, {} needed a second attempt".format(fails, N, retried))
+    return fails
+
+
+sys.exit(1 if asyncio.run(main()) else 0)

tests/bledev_board/repl_client.py

@@ -7,6 +7,7 @@
 board (mpble); see gatt_central.py for the command line. Checks, in order:
 
 1. The right password gets a prompt, and ``123 * 456`` evaluates to 56088.
+   A 16,890-character print arrives byte for byte.
 2. A wrong password, sent together with a line of code that would create
    ``/pwned``, gets ``Access denied`` and a disconnect, and never a prompt.
 3. Logged in again, ``/pwned`` does not exist: the code never ran.
@@ -65,6 +66,11 @@ async def main():
     link = await repl.connect(ble, PASSWORD, name=NAME, timeout_ms=15000)
     out = await repl.run(link, "123 * 456")
     check(out.strip() == "56088", "the right password gets a working prompt", repr(out))
+    # A list prints as hundreds of small writes, which is what once let the
+    # retry timer interleave two flushes: a chunk sent twice, the next lost.
+    out = await repl.run(link, "print(list(range(3000)))", timeout_ms=30000)
+    want = str(list(range(3000)))
+    check(out.strip() == want, "a long print arrives intact", "{} chars, want {}".format(len(out.strip()), len(want)))
     await repl.run(link, "import os")
     await repl.run(link, "exec(\"try:\\n os.remove('/pwned')\\nexcept OSError:\\n pass\")")
     out = await repl.run(link, "'pwned' in os.listdir('/')")