fix(micropython): release the raw REPL's answer instead of buffering it (#219)

MicroPython over the QEMU worker stopped loading: every project failed with
"the board stopped answering while writing _vlx_main.py", the board sitting
idle at a healthy `>>>`.

The uploader shipped in 664c3581 waits for the board's end-of-execution marker
before sending the next step — that wait is the whole point, it replaced a
fixed delay that pushed 64 bytes every 150 ms. The marker never arrived. Raw
REPL answers each statement with `OK` <EOT> traceback <EOT> `>` and not one
newline in it, and the UART buffer flushed only on newline, carriage return, a
period or 256 bytes. The answer sat in the buffer; the uploader timed out
waiting for something the backend was holding.

Nothing had needed that answer before: the old uploader pasted the whole
program and never listened. EOT is a message boundary in this protocol, so the
buffer treats it as one.

Found by forcing a MicroPython project down the QEMU path with ?esp32sim=qemu —
the residual risk 664c3581 flagged and could not exercise, because velxio.dev
routes every supported ESP32 kind to the in-browser engines. Self-hosters from
this repo have no engines and always take this path, so it broke for exactly
the people the #219 fix was written for.
This commit is contained in:
David Montero Crespo 2026-08-21 22:06:16 +02:00
parent 5b277e85e4
commit 89025212ca
2 changed files with 56 additions and 3 deletions

View File

@ -152,9 +152,16 @@ class _UartBuffer:
"""Add one byte. Returns decoded string when a flush occurs, else None."""
with self._lock:
self._buf.append(byte_val)
# Flush on newline, carriage return, period, or max size
# This ensures progress dots '...' don't buffer endlessly.
if byte_val in (ord('\n'), ord('\r'), ord('.')) or len(self._buf) >= self.flush_size:
# Flush on newline, carriage return, period, EOT, or max size.
# The dots keep a progress line from buffering endlessly; EOT
# (Ctrl-D, 0x04) is there because MicroPython's raw REPL answers
# every statement with `OK` <EOT> traceback <EOT> `>` and not one
# newline in it. Held here, that answer never reaches the uploader
# waiting on it, and a MicroPython project times out mid-upload
# with the board sitting idle at the prompt. It only surfaced once
# the uploader started waiting for the board instead of pasting
# blind (see frontend simulation/micropythonSession.ts).
if byte_val in (ord('\n'), ord('\r'), ord('.'), 0x04) or len(self._buf) >= self.flush_size:
text = self._buf.decode('utf-8', errors='replace')
self._buf.clear()
return text

View File

@ -0,0 +1,46 @@
"""
When the UART buffer lets go of what it has (issue #260 verification).
MicroPython's raw REPL answers every statement with `OK` <EOT> traceback <EOT>
`>` and not a single newline. The buffer used to flush only on newline,
carriage return, a period or 256 bytes, so that answer sat in it — and the
project uploader, which waits for the board's end-of-execution marker before
sending the next step, timed out with the board idle at the prompt.
"""
import unittest
from app.services.esp32_lib_manager import _UartBuffer
def feed_all(buf: _UartBuffer, text: bytes) -> str:
"""Everything the buffer released while consuming `text`."""
return ''.join(out for b in text if (out := buf.feed(b)) is not None)
class TestUartBufferFlush(unittest.TestCase):
def test_raw_repl_answer_is_released(self):
out = feed_all(_UartBuffer(0), b'OK\x04\x04>')
self.assertEqual(out.count('\x04'), 2)
def test_answer_carrying_a_traceback_is_released(self):
out = feed_all(_UartBuffer(0), b'OK\x04OSError: 28\x04>')
self.assertEqual(out.count('\x04'), 2)
self.assertIn('OSError: 28', out)
def test_the_trailing_prompt_still_waits(self):
"""`>` is not a boundary; it rides out with the next flush. The
uploader keys on the EOT pair, so this is fine."""
buf = _UartBuffer(0)
out = feed_all(buf, b'OK\x04\x04>')
self.assertFalse(out.endswith('>'))
self.assertEqual(buf.flush(), '>')
def test_ordinary_lines_still_flush_on_newline(self):
self.assertEqual(feed_all(_UartBuffer(0), b'hello\n'), 'hello\n')
def test_progress_dots_still_flush(self):
self.assertEqual(feed_all(_UartBuffer(0), b'...'), '...')
def test_size_limit_still_applies(self):
buf = _UartBuffer(0, flush_size=8)
self.assertEqual(feed_all(buf, b'abcdefgh'), 'abcdefgh')