From 7db89231b6ae4904a9d26ad57cf30b7010969c47 Mon Sep 17 00:00:00 2001 From: MCKero Date: Fri, 28 Aug 2026 09:59:34 +0100 Subject: [PATCH] Add a bounded log buffer Collects power events, QEMU stderr, and firmware serial output. Bounded and in memory on purpose: an unbounded buffer in a long-running server is a slow leak, and anyone wanting a permanent record can redirect the server's stderr. Entries carry a monotonic seq so a polling client can ask for "anything after N" and receive each line exactly once, including after eviction has dropped older entries -- a test covers that case specifically, since an index-based cursor would silently repeat or skip lines there. Stream decoding is lenient: serial bytes can be garbage before the firmware configures the port, and losing the stream to one bad byte would be worse than a replacement character. --- tools/test_uvk5_logs.py | 94 +++++++++++++++++++++++++++++++++++++++++ tools/uvk5_logs.py | 61 ++++++++++++++++++++++++++ 2 files changed, 155 insertions(+) create mode 100644 tools/test_uvk5_logs.py create mode 100644 tools/uvk5_logs.py diff --git a/tools/test_uvk5_logs.py b/tools/test_uvk5_logs.py new file mode 100644 index 0000000..1ea0e3f --- /dev/null +++ b/tools/test_uvk5_logs.py @@ -0,0 +1,94 @@ +#!/usr/bin/env python3 +"""Unit tests for the log ring buffer. No emulator needed.""" +import io +import threading +import unittest + +from uvk5_logs import LogBuffer + + +class TestLogBuffer(unittest.TestCase): + def test_records_and_returns_entries(self): + log = LogBuffer(capacity=10) + log.add("power", "on") + entries = log.entries() + self.assertEqual(len(entries), 1) + self.assertEqual(entries[0]["source"], "power") + self.assertEqual(entries[0]["text"], "on") + self.assertIn("time", entries[0]) + self.assertIn("seq", entries[0]) + + def test_is_bounded(self): + """Unbounded growth inside a long-running server is a slow leak.""" + log = LogBuffer(capacity=5) + for i in range(20): + log.add("test", f"line {i}") + entries = log.entries() + self.assertEqual(len(entries), 5) + self.assertEqual(entries[-1]["text"], "line 19") + self.assertEqual(entries[0]["text"], "line 15") + + def test_since_returns_only_newer_entries(self): + log = LogBuffer(capacity=10) + log.add("a", "first") + cursor = log.cursor() + log.add("b", "second") + fresh = log.entries(since=cursor) + self.assertEqual([e["text"] for e in fresh], ["second"]) + + def test_since_beyond_the_end_returns_nothing(self): + log = LogBuffer(capacity=10) + log.add("a", "first") + self.assertEqual(log.entries(since=log.cursor()), []) + + def test_cursor_survives_eviction(self): + """A client polling with `since` must not be sent the same line twice + just because older entries were dropped.""" + log = LogBuffer(capacity=3) + for i in range(3): + log.add("t", f"{i}") + cursor = log.cursor() + for i in range(3, 6): + log.add("t", f"{i}") + fresh = log.entries(since=cursor) + self.assertEqual([e["text"] for e in fresh], ["3", "4", "5"]) + + def test_pump_stream_splits_lines_and_tags_serial(self): + log = LogBuffer(capacity=20) + stream = io.BytesIO(b"SERIAL boot ok\nplain qemu message\n") + log.pump_stream(stream, default_source="qemu") + got = [(e["source"], e["text"]) for e in log.entries()] + self.assertIn(("serial", "boot ok"), got) + self.assertIn(("qemu", "plain qemu message"), got) + + def test_pump_stream_skips_blank_lines(self): + log = LogBuffer(capacity=20) + log.pump_stream(io.BytesIO(b"one\n\n\ntwo\n"), default_source="qemu") + self.assertEqual([e["text"] for e in log.entries()], ["one", "two"]) + + def test_pump_stream_survives_undecodable_bytes(self): + """Serial output can be garbage before the firmware initialises.""" + log = LogBuffer(capacity=20) + log.pump_stream(io.BytesIO(b"\xff\xfe bad\ngood\n"), default_source="qemu") + self.assertEqual(len(log.entries()), 2) + + def test_add_is_thread_safe(self): + log = LogBuffer(capacity=500) + + def worker(n): + for i in range(100): + log.add("t", f"{n}-{i}") + + threads = [threading.Thread(target=worker, args=(n,)) for n in range(4)] + for t in threads: + t.start() + for t in threads: + t.join() + self.assertEqual(len(log.entries()), 400) + # sequence numbers must be unique, or `since` would skip or repeat lines + seqs = [e["seq"] for e in log.entries()] + self.assertEqual(len(seqs), len(set(seqs))) + + +if __name__ == "__main__": + unittest.main() diff --git a/tools/uvk5_logs.py b/tools/uvk5_logs.py new file mode 100644 index 0000000..1e24ffd --- /dev/null +++ b/tools/uvk5_logs.py @@ -0,0 +1,61 @@ +#!/usr/bin/env python3 +"""A bounded log buffer for the web UI. + +Collects supervisor events, QEMU stderr, and firmware serial output (which the +machine model prints as "SERIAL ") so the browser has something to show. + +In memory and bounded on purpose: this is a debugging aid inside a long-running +server, so an unbounded buffer would be a slow leak. Anyone wanting a permanent +record can redirect the server's own stderr to a file. + +Every entry carries a monotonic `seq`, so a polling client can ask for "anything +after N" and get each line exactly once even when older entries have been evicted. +""" +import collections +import threading +import time + + +class LogBuffer: + def __init__(self, capacity: int = 500): + self._entries = collections.deque(maxlen=capacity) + self._lock = threading.Lock() + self._seq = 0 + + def add(self, source: str, text: str): + with self._lock: + self._seq += 1 + self._entries.append({ + "seq": self._seq, + "time": time.strftime("%H:%M:%S"), + "source": source, + "text": text, + }) + + def cursor(self) -> int: + with self._lock: + return self._seq + + def entries(self, since: int = 0): + with self._lock: + return [e for e in self._entries if e["seq"] > since] + + def pump_stream(self, stream, default_source: str = "qemu"): + """Read a byte stream to EOF, one entry per line. + + Lines the machine model tags with "SERIAL " are firmware output and are + recorded under their own source, so the UI can tell them apart from QEMU's + own chatter. + + Decoding is lenient: serial bytes can be garbage before the firmware has + configured the port, and losing the whole stream to one bad byte would be + worse than showing a replacement character. + """ + for raw in iter(stream.readline, b""): + line = raw.decode("utf-8", "replace").rstrip("\r\n") + if not line: + continue + if line.startswith("SERIAL "): + self.add("serial", line[len("SERIAL "):]) + else: + self.add(default_source, line)