From b226229c747e0dad6bc2cb3b9ca0cd60f9b7c6c4 Mon Sep 17 00:00:00 2001 From: Brad Barnett <127794626+bdbarnett@users.noreply.github.com> Date: Mon, 9 Nov 2026 20:03:55 -0600 Subject: [PATCH] romfs deploy: works over WebREPL while a program runs from /rom Over WebREPL there's no soft reset before a deploy, so a program imported from /rom kept running between the deploy's commands. Its timers ran bytecode the deploy had just erased, and the names it had brought in read back as erased flash, so the board hung with a blank partition. A deploy that sends one command per 4 KB chunk is also left half-written, under a valid header, when the link drops midway. Over WebREPL and BLE, and over serial when modules from /rom are still loaded, the erase and write now run on the board inside one scheduled callback, which keeps every other callback out. The image streams through it, the header chunk goes last, and the board resets when it's done, or when the host gives up and sends Ctrl-C. On an Arduino GIGA R1 running its launcher from /rom, two WebREPL deploys of a 1,883,072-byte image finished in 50 s and 49 s, the image read back identical and the board came back up running the launcher from it. Fixes #215 --- cli/src/mpftp/cli.py | 3 + cli/src/mpftp/sidecar.py | 233 +++++++++++++++++++++ cli/tests/test_romfs_deploy_running.py | 275 +++++++++++++++++++++++++ docs/reference/cli.md | 3 +- 4 files changed, 513 insertions(+), 1 deletion(-) create mode 100644 cli/tests/test_romfs_deploy_running.py diff --git a/cli/src/mpftp/cli.py b/cli/src/mpftp/cli.py index 60c899a..9ae201b 100755 --- a/cli/src/mpftp/cli.py +++ b/cli/src/mpftp/cli.py @@ -1325,6 +1325,9 @@ def cmd_romfs(ns: argparse.Namespace) -> None: if ns.romfs_cmd == "query": out(client.call("romfs_query")) elif ns.romfs_cmd == "deploy": + # Erasing and writing a few megabytes over Wi-Fi outlasts the call timeout. + if isinstance(client, TcpClient): + client.call_timeout = None out( client.call( "romfs_deploy", diff --git a/cli/src/mpftp/sidecar.py b/cli/src/mpftp/sidecar.py index 45a7c01..b3a7ec7 100644 --- a/cli/src/mpftp/sidecar.py +++ b/cli/src/mpftp/sidecar.py @@ -27,6 +27,7 @@ import threading import time import traceback +import zlib from pathlib import Path from typing import Any, Callable, Optional @@ -677,6 +678,129 @@ def _error( _emit(err) +# The board side of a ROMFS deploy that can't soft-reset first. It runs as one +# scheduled callback: while it runs, no other callback can, so nothing loaded +# from the image runs while it's erased and rewritten. It reports E per erase +# step, R when ready, W per chunk written, D when done or X and the error, then +# resets the board. The markers are bytes made before the erase: a one-letter +# string is an interned name, whose text may live in the image being erased. +_ROMFS_DEPLOY_SCRIPT = """\ +import sys,vfs,machine,micropython,time +from binascii import a2b_base64 +_Z={deflate} +if _Z: + from io import BytesIO + from deflate import DeflateIO,RAW +_M=bytearray(b'ERWDX\\n') +def _mpftp_romfs(_): + i=sys.stdin.buffer + o=sys.stdout.buffer + try: + try: + vfs.umount('/rom') + except OSError: + pass + dev=vfs.rom_ioctl(2,{rom_id}) + off=0 + while off<{length}: + if {has_object}: + dev.ioctl(6,off//{block}) + off+={block} + else: + n=(min({length}-off,32768)+{block}-1)//{block}*{block} + r=vfs.rom_ioctl(3,{rom_id},off,n) + if r<0: + raise OSError(-r) + off+=n + o.write(_M[0:1]) + o.write(_M[1:2]) + while True: + h=i.read(16) + off=int(h[:8],16) + n=int(h[8:],16) + if not n: + break + b=a2b_base64(i.read(n)) + if _Z: + b=DeflateIO(BytesIO(b),RAW,9).read() + if {has_object}: + dev.writeblocks(off//{block},b,off%{block}) + else: + r=vfs.rom_ioctl(4,{rom_id},off,b) + if r<0: + raise OSError(-r) + o.write(_M[2:3]) + if not {has_object}: + vfs.rom_ioctl(5,{rom_id}) + o.write(_M[3:4]) + except Exception as e: + o.write(_M[4:5]) + try: + o.write(repr(e).encode()) + except Exception: + pass + o.write(_M[5:6]) + finally: + time.sleep_ms(300) + machine.reset() +while True: + try: + micropython.schedule(_mpftp_romfs,None) + break + except RuntimeError: + time.sleep_ms(1) +while True: + pass +""" + +# Modules whose code came from a ROMFS mount: their bytecode runs in place. +_ROMFS_LOADED_EXPR = ( + "[k for k,m in __import__('sys').modules.items() " + "if str(getattr(m,'__file__','')).startswith('/rom/')]" +) + + +def _romfs_modules_loaded(transport: Any) -> list[str]: + """Names of the modules the board imported from ``/rom``.""" + try: + return list(transport.eval(_ROMFS_LOADED_EXPR)) + except Exception: + return [] + + +def _romfs_read_marker(serial: Any, want: bytes, stall_secs: float) -> bytes: + """Read the deploy script's progress until ``want``; X carries an error.""" + seen = b"" + last = time.monotonic() + while True: + if serial.inWaiting() > 0: + c = serial.read(1) + last = time.monotonic() + if c == want: + return c + if c == b"X": + line = b"" + while not line.endswith(b"\n") and time.monotonic() - last < stall_secs: + if serial.inWaiting() > 0: + line += serial.read(1) + else: + time.sleep(0.005) + raise RuntimeError( + "the board failed during the ROMFS deploy and reset: " + + line.decode("utf-8", "replace").strip() + + "; the partition holds no valid image until a deploy completes" + ) + seen = (seen + c)[-64:] + elif time.monotonic() - last > stall_secs: + raise RuntimeError( + f"the board went quiet for {stall_secs:.0f} s during the ROMFS deploy " + f"(waiting for {want.decode()}, last saw {seen!r}); it may need a reset, " + "and the partition holds no valid image until a deploy completes" + ) + else: + time.sleep(0.002) + + def _error_from_exception(req_id: Any, exc: BaseException) -> None: """Error reply for an exception a METHODS handler raised. Call it from the ``except`` block so the traceback is still the current one.""" @@ -4255,6 +4379,13 @@ def romfs_deploy( with self._lock: self._stop_repl_reader() state = self._mp_state() + state.ensure_raw_repl() + running = _romfs_modules_loaded(state.transport) + # Over the air there was no soft reset, and a dropped link midway + # would leave a half-written image under a valid header, so the + # wireless links always take the one-step deploy. + if running or self.is_network or self.is_ble: + return self._romfs_deploy_and_reset(path, partition, mpy, running) args = argparse.Namespace(path=path, partition=partition, mpy=mpy) buf = io.StringIO() with contextlib.redirect_stdout(buf): @@ -4268,6 +4399,108 @@ def romfs_deploy( self._start_repl_reader() return {"output": buf.getvalue().strip(), "path": path, "partition": partition} + #: Longest the board may go quiet during a deploy: one erase step or chunk. + ROMFS_STALL_SECS = 30.0 + + def _romfs_deploy_and_reset( + self, path: str, partition: int, mpy: bool, running: list[str] + ) -> dict[str, Any]: + """Deploy in one step on the board, then reset it. + + A program imported from ``/rom`` runs its bytecode in place, and the + names it brought in keep their text there, so its timers and + callbacks would run erased or rewritten flash between the deploy's + steps, and the board hangs. A connection that soft-resets first + (serial) clears them; WebREPL and BLE can't, since a soft reset drops + the link. So the whole erase and write runs inside one scheduled + callback, which keeps every other callback from running, and the + board resets at the end, because nothing loaded from the old image + can run again. The image streams through that one call instead of a + command per chunk, and the first chunk, which holds the header, goes + last, so a deploy that stops early leaves no valid image to mount. + """ + from mpremote.romfs import VfsRomWriter, make_romfs + + t = self._require() + if os.path.isfile(path) and path.endswith((".img", ".romfs")): + romfs = Path(path).read_bytes() + else: + romfs = make_romfs(path, mpy_cross=mpy) + if not romfs.startswith(VfsRomWriter.ROMFS_HEADER): + raise RuntimeError(f"{path}: not a ROMFS image") + # The partition, as mpremote's own deploy finds it. + t.exec("import vfs") + if not t.eval("hasattr(vfs,'rom_ioctl')"): + raise RuntimeError("ROMFS is not enabled on this device") + t.exec(f"dev=vfs.rom_ioctl(2,{partition})") + if t.eval("isinstance(dev,int) and dev<0"): + raise RuntimeError(f"ROMFS{partition} partition not found on device") + has_object = t.eval("hasattr(dev,'ioctl')") + if has_object: + block = t.eval("dev.ioctl(5,0)") + rom_size = t.eval("dev.ioctl(4,0)") * block + else: + rom_size = t.eval("len(dev)") + block = t.eval(f"vfs.rom_ioctl(6,{partition})") + if block <= 0: + block = rom_size + if len(romfs) > rom_size: + raise RuntimeError( + f"the image ({len(romfs)} bytes) is too big for ROMFS{partition} ({rom_size} bytes)" + ) + try: + t.exec("from io import BytesIO\nfrom deflate import DeflateIO,RAW") + deflate = True + except Exception: + deflate = False + chunk = min(4096, block) if has_object else 4096 + offsets = list(range(chunk, len(romfs), chunk)) + [0] + + t.exec_raw_no_follow( + _ROMFS_DEPLOY_SCRIPT.format( + deflate=int(deflate), + rom_id=partition, + length=len(romfs), + block=block, + has_object=int(has_object), + ) + ) + serial = t.serial + result: dict[str, Any] = { + "path": path, + "partition": partition, + "size": len(romfs), + "loaded_from_rom": running, + "reset": True, + } + try: + _romfs_read_marker(serial, b"R", self.ROMFS_STALL_SECS) + for offset in offsets: + data = romfs[offset : offset + chunk] + data += bytes(chunk - len(data)) + if deflate: + z = zlib.compressobj(wbits=-9) + data = z.compress(data) + z.flush() + b64 = base64.b64encode(data) + serial.write(b"%08x%08x" % (offset, len(b64)) + b64) + _romfs_read_marker(serial, b"W", self.ROMFS_STALL_SECS) + serial.write(b"%08x%08x" % (0, 0)) + _romfs_read_marker(serial, b"D", self.ROMFS_STALL_SECS) + except RuntimeError as e: + # A board left waiting for data would hold its scheduler: Ctrl-C + # ends the wait, and the script resets on the way out. + try: + serial.write(b"\x03\x03") + except Exception: + pass + result.update({"ok": False, "error": str(e)}) + else: + result.update({"ok": True, "output": f"ROMFS image deployed ({len(romfs)} bytes)"}) + finally: + self._force_close_transport(graceful=False) + result["note"] = "the board resets after the deploy; reconnect when it's back" + return result + def fs_cp(self, src: str, dest: str, verify: bool = False, mpy: bool = False) -> dict[str, Any]: """ Recursive copy using mpremote ':' prefix for board paths. diff --git a/cli/tests/test_romfs_deploy_running.py b/cli/tests/test_romfs_deploy_running.py new file mode 100644 index 0000000..48a9bed --- /dev/null +++ b/cli/tests/test_romfs_deploy_running.py @@ -0,0 +1,275 @@ +# SPDX-FileCopyrightText: 2026 Brad Barnett +# +# SPDX-License-Identifier: MIT + +"""romfs deploy while modules from /rom are loaded: one scheduled step, then a reset. + +The board side runs here under CPython against a fake partition, through a +fake serial link, so the host and board halves of the protocol are tested +together. +""" + +from __future__ import annotations + +import builtins +import io +import queue +import tempfile +import threading +import time +import types +import unittest +import zlib +from pathlib import Path +from unittest import mock + +from mpftp import sidecar +from mpftp.sidecar import Session + +ROM_SIZE = 64 * 1024 +BLOCK = 4096 + + +class _Reset(BaseException): + pass + + +class FakeBoard: + """A ROMFS partition, the modules the deploy script imports, and a link.""" + + def __init__(self, fail_write_at=None): + self.rom = bytearray(b"\x5a" * ROM_SIZE) + self.erased = [] + self.writes = [] + self.reset = threading.Event() + self.fail_write_at = fail_write_at + self.to_board = queue.Queue() + self.from_board = bytearray() + self.lock = threading.Lock() + + # The board's stdin and stdout. + def board_read(self, n): + out = b"" + while len(out) < n: + out += self.to_board.get(timeout=5) + extra = out[n:] + if extra: + # Put back what this read didn't ask for, ahead of anything newer. + rest = [extra] + while not self.to_board.empty(): + rest.append(self.to_board.get()) + for r in rest: + self.to_board.put(r) + return out[:n] + + def board_write(self, s): + with self.lock: + self.from_board += s.encode() if isinstance(s, str) else s + + def rom_ioctl(self, cmd, rom_id, *args): + if cmd == 2: + return memoryview(self.rom) + if cmd == 3: + off, n = args + self.rom[off : off + n] = b"\xff" * n + self.erased.append((off, n)) + return 4 + if cmd == 4: + off, data = args + if self.fail_write_at is not None and off == self.fail_write_at: + return -5 + self.writes.append(off) + self.rom[off : off + len(data)] = data + return 0 + if cmd == 5: + return 0 + if cmd == 6: + return BLOCK + raise AssertionError(cmd) + + def modules(self): + board = self + + class _Stdin: + buffer = types.SimpleNamespace(read=board_read_wrapper(board)) + + def umount(_p): + raise OSError(22) + + def reset(): + board.reset.set() + raise _Reset + + def schedule(fn, arg): + fn(arg) + + class DeflateIO: + def __init__(self, stream, _fmt, _wbits): + self.stream = stream + + def read(self): + return zlib.decompress(self.stream.read(), wbits=-9) + + return { + "sys": types.SimpleNamespace( + stdin=_Stdin, + stdout=types.SimpleNamespace(buffer=types.SimpleNamespace(write=board.board_write)), + ), + "vfs": types.SimpleNamespace(rom_ioctl=board.rom_ioctl, umount=umount), + "machine": types.SimpleNamespace(reset=reset), + "micropython": types.SimpleNamespace(schedule=schedule), + "time": types.SimpleNamespace(sleep_ms=lambda _ms: None), + "binascii": __import__("binascii"), + "io": types.SimpleNamespace(BytesIO=io.BytesIO), + "deflate": types.SimpleNamespace(DeflateIO=DeflateIO, RAW=0), + } + + def run_script(self, script): + mods = self.modules() + real_import = builtins.__import__ + + def fake_import(name, globals=None, locals=None, fromlist=(), level=0): + if name in mods: + return mods[name] + return real_import(name, globals, locals, fromlist, level) + + env = {"__builtins__": dict(vars(builtins), __import__=fake_import)} + + def go(): + try: + exec(script, env) + except _Reset: + pass + + threading.Thread(target=go, daemon=True).start() + + +def board_read_wrapper(board): + return lambda n: board.board_read(n) + + +class FakeSerial: + def __init__(self, board): + self.board = board + + def write(self, data): + self.board.to_board.put(bytes(data)) + return len(data) + + def inWaiting(self): # noqa: N802 + with self.board.lock: + return len(self.board.from_board) + + def read(self, n=1): + with self.board.lock: + out = bytes(self.board.from_board[:n]) + del self.board.from_board[:n] + return out + + +class FakeTransport: + def __init__(self, board): + self.board = board + self.serial = FakeSerial(board) + self.in_raw_repl = True + + def exec(self, code): + return b"" + + def eval(self, expr): + answers = { + "hasattr(vfs,'rom_ioctl')": True, + "isinstance(dev,int) and dev<0": False, + "hasattr(dev,'ioctl')": False, + "len(dev)": ROM_SIZE, + "vfs.rom_ioctl(6,0)": BLOCK, + } + return answers[expr] + + def exec_raw_no_follow(self, script): + self.board.run_script(script) + + +def make_image(size): + body = bytes((i * 7) & 0xFF for i in range(size - 3)) + return b"\xd2\xcd\x31" + body + + +class DeployWhileRunningTests(unittest.TestCase): + def deploy(self, board, image): + session = Session() + session.transport = FakeTransport(board) + closed = [] + with tempfile.TemporaryDirectory() as td, mock.patch.object( + Session, "_require", lambda self: self.transport + ), mock.patch.object( + Session, "_force_close_transport", lambda self, graceful=False: closed.append(graceful) + ): + path = Path(td) / "app.romfs" + path.write_bytes(image) + result = session._romfs_deploy_and_reset(str(path), 0, True, ["appkit"]) + self.assertTrue(board.reset.wait(5), "the board was never reset") + self.assertEqual(closed, [False]) + return result + + def test_writes_the_image_header_last_and_resets(self): + board = FakeBoard() + image = make_image(3 * BLOCK + 100) + result = self.deploy(board, image) + self.assertTrue(result["ok"], result) + self.assertEqual(bytes(board.rom[: len(image)]), image) + self.assertEqual(board.writes[-1], 0) + self.assertEqual(sorted(board.writes), [0, BLOCK, 2 * BLOCK, 3 * BLOCK]) + self.assertEqual(board.erased, [(0, 4 * BLOCK)]) + + def test_a_failed_write_is_reported_and_still_resets(self): + board = FakeBoard(fail_write_at=BLOCK) + result = self.deploy(board, make_image(2 * BLOCK)) + self.assertFalse(result["ok"]) + self.assertIn("OSError", result["error"]) + # The header chunk never went in, so nothing valid is left to mount. + self.assertNotIn(0, board.writes) + + def test_deploy_takes_this_path_only_with_modules_from_rom(self): + session = Session() + session.interpreter = "micropython" + transport = mock.Mock(in_raw_repl=True) + transport.eval.return_value = ["appkit", "events"] + session.transport = transport + with mock.patch.object(Session, "_require", lambda self: transport), mock.patch.object( + Session, "_romfs_deploy_and_reset", return_value={"ok": True} + ) as locked, mock.patch.object(Session, "_require_micropython", lambda self, what: None): + self.assertEqual(session.romfs_deploy("x.romfs"), {"ok": True}) + locked.assert_called_once_with("x.romfs", 0, True, ["appkit", "events"]) + + def test_wifi_always_takes_this_path(self): + session = Session() + session.interpreter = "micropython" + session.device = "ws://192.168.1.50" + transport = mock.Mock(in_raw_repl=True) + transport.eval.return_value = [] + session.transport = transport + with mock.patch.object(Session, "_require", lambda self: transport), mock.patch.object( + Session, "_romfs_deploy_and_reset", return_value={"ok": True} + ) as locked, mock.patch.object(Session, "_require_micropython", lambda self, what: None): + session.romfs_deploy("x.romfs") + locked.assert_called_once_with("x.romfs", 0, True, []) + + +class MarkerReaderTests(unittest.TestCase): + def test_a_quiet_board_is_reported(self): + board = FakeBoard() + with self.assertRaisesRegex(RuntimeError, "went quiet"): + sidecar._romfs_read_marker(FakeSerial(board), b"R", 0.05) + + def test_an_error_line_is_reported(self): + board = FakeBoard() + board.board_write("EEXOSError(5,)\n") + start = time.monotonic() + with self.assertRaisesRegex(RuntimeError, r"OSError\(5,\)"): + sidecar._romfs_read_marker(FakeSerial(board), b"R", 5) + self.assertLess(time.monotonic() - start, 1) + + +if __name__ == "__main__": + unittest.main() diff --git a/docs/reference/cli.md b/docs/reference/cli.md index f4afa31..914cc9c 100644 --- a/docs/reference/cli.md +++ b/docs/reference/cli.md @@ -29,7 +29,8 @@ and exits non-zero. | `hash`, `mkdir`, `rm`, `touch`, `rename` | As named. `rm -r` removes a tree | | `edit` | Edit a board file with `$EDITOR` | | `clean` | Remove debris a session leaves (probe scratch files, `.bak`, `__pycache__`) | -| `mount`, `umount`, `romfs` | MicroPython over serial only | +| `mount`, `umount` | MicroPython over serial only | +| `romfs query`, `romfs build DIR`, `romfs deploy PATH` | Read, build or write a MicroPython ROMFS image. Over WebREPL, `deploy` writes the image in one step and resets the board when it's done, so a program running from `/rom` can't run while its image is rewritten | ## Run and reset