From 7d087f2a413b4a903fedb75318872326a1e9ff89 Mon Sep 17 00:00:00 2001 From: jay23606 <4602056+jay23606@users.noreply.github.com> Date: Fri, 16 Oct 2026 16:15:51 -0400 Subject: [PATCH 1/4] microvm tests: stream OpenVMM transcripts to disk as they are read OpenvmmProcess kept the transcript in memory and wrote it only from wait or close, so a run terminated before cleanup left no transcript, or a truncated one, even though output had been read. Open the transcript before OpenVMM starts, append each consumed chunk and flush it, and close the file at wait or close. Bytes are written in the order received, with no prefixes and no duplication at cleanup. An error creating the transcript stops the run before OpenVMM starts. Output still queued at close is written too. The in-memory output and marker search are unchanged. Adds tests for growth while the process runs, a transcript that cannot be created, and output delivered after the last wait. Fixes #176 Co-Authored-By: Claude Haiku 5.5 Signed-off-by: jay23606 <4602056+jay23606@users.noreply.github.com> --- scripts/nvx_tools/openvmm_process.py | 34 ++++++++++++++++------ scripts/test_microvm_tests.py | 42 ++++++++++++++++++++++++++++ 2 files changed, 67 insertions(+), 9 deletions(-) diff --git a/scripts/nvx_tools/openvmm_process.py b/scripts/nvx_tools/openvmm_process.py index 7e4184587..c7f18cf8c 100644 --- a/scripts/nvx_tools/openvmm_process.py +++ b/scripts/nvx_tools/openvmm_process.py @@ -83,7 +83,15 @@ def __init__( process_environment["OPENVMM_LOG"] = "off" if environment is not None: process_environment.update(environment) - self._interaction = InteractiveProcess(command, process_environment) + # The transcript is open before OpenVMM starts, so a failure to create it + # stops the run here, and every consumed byte reaches it as it is read. + log_path.parent.mkdir(parents=True, exist_ok=True) + self._log = log_path.open("wb") + try: + self._interaction = InteractiveProcess(command, process_environment) + except BaseException: + self._log.close() + raise self._chunks: queue.Queue[bytes | None] = queue.Queue() self._reader = threading.Thread( target=self._interaction.read_output, @@ -97,6 +105,7 @@ def __init__( except BaseException: terminate(self._interaction.process) self._interaction.close() + self._log.close() raise self._output = bytearray() self._search_offset = 0 @@ -154,7 +163,7 @@ def _query_status_first(self) -> None: self._consume(chunk) def _consume(self, chunk: bytes) -> None: - self._output.extend(chunk) + self._record(chunk) try: self._monitor.feed(chunk) except TimeAbiFailure as error: @@ -266,7 +275,7 @@ def wait(self, timeout: float) -> OpenvmmProcessResult: except TimeAbiFailure as error: self._fail(error) self._finished = True - self._write_log() + self._close_log() return OpenvmmProcessResult(returncode, bytes(self._output)) def close(self) -> None: @@ -279,8 +288,9 @@ def close(self) -> None: """ if not self._finished and self.process.poll() is None: terminate(self.process) - self._drain_available() - self._write_log() + if not self._log.closed: + self._drain_available() + self._close_log() self._interaction.close() self._finished = True @@ -292,11 +302,17 @@ def _drain_available(self) -> None: except queue.Empty: return if chunk is not None: - self._output.extend(chunk) + self._record(chunk) + + def _record(self, chunk: bytes) -> None: + """Keep a consumed chunk in memory and in the transcript, flushed for readers.""" + self._output.extend(chunk) + self._log.write(chunk) + self._log.flush() - def _write_log(self) -> None: - self._log_path.parent.mkdir(parents=True, exist_ok=True) - self._log_path.write_bytes(self._output) + def _close_log(self) -> None: + if not self._log.closed: + self._log.close() def _fail(self, error: Exception): self.close() diff --git a/scripts/test_microvm_tests.py b/scripts/test_microvm_tests.py index 6a0733642..660737812 100644 --- a/scripts/test_microvm_tests.py +++ b/scripts/test_microvm_tests.py @@ -4827,6 +4827,48 @@ def test_process_wait_reads_final_chunks_after_process_exit(self): self.assertEqual(log_path.read_bytes(), result.output) self.assertEqual(queues.return_value.get.call_count, 4) + def test_process_transcript_grows_while_the_process_runs(self): + with tempfile.TemporaryDirectory() as temporary: + log_path = Path(temporary) / "output.log" + with ( + patch.object(openvmm_process, "InteractiveProcess") as interaction, + patch.object(openvmm_process.threading, "Thread"), + patch.object(openvmm_process.queue, "Queue") as queues, + ): + interaction.return_value.process.poll.return_value = None + interaction.return_value.process.wait.return_value = 0 + queues.return_value.get.side_effect = [b"BOOT\n", None] + queues.return_value.get_nowait.side_effect = queue.Empty + with openvmm_process.OpenvmmProcess(["openvmm"], log_path) as process: + process.wait_for(b"BOOT", 1) + # Consumed bytes are on disk while OpenVMM is still running. + self.assertEqual(log_path.read_bytes(), b"BOOT\n") + process.wait(1) + self.assertEqual(log_path.read_bytes(), b"BOOT\n") + + def test_process_transcript_open_failure_surfaces_before_start(self): + with tempfile.TemporaryDirectory() as temporary: + log_path = Path(temporary) / "output.log" + log_path.mkdir() + with patch.object(openvmm_process, "InteractiveProcess") as interaction: + with self.assertRaises(OSError): + openvmm_process.OpenvmmProcess(["openvmm"], log_path) + interaction.assert_not_called() + + def test_process_close_writes_output_delivered_after_the_last_wait(self): + with tempfile.TemporaryDirectory() as temporary: + log_path = Path(temporary) / "output.log" + with ( + patch.object(openvmm_process, "InteractiveProcess") as interaction, + patch.object(openvmm_process.threading, "Thread"), + patch.object(openvmm_process.queue, "Queue") as queues, + ): + interaction.return_value.process.poll.return_value = None + queues.return_value.get_nowait.side_effect = [b"LATE\n", queue.Empty] + process = openvmm_process.OpenvmmProcess(["openvmm"], log_path) + process.close() + self.assertEqual(log_path.read_bytes(), b"LATE\n") + def test_process_wait_clamps_elapsed_process_timeout(self): with tempfile.TemporaryDirectory() as temporary: with ( From 20cbf65b3fdde4e2b2980f6f158f14ae76cf985a Mon Sep 17 00:00:00 2001 From: jay23606 <4602056+jay23606@users.noreply.github.com> Date: Fri, 16 Oct 2026 16:32:35 -0400 Subject: [PATCH 2/4] microvm tests: drain queued output after interaction close and always release the transcript close drained the reader's queue before the interaction had stopped, so a final chunk queued during shutdown was lost from both output and transcript. A write failure during that drain also skipped closing the interaction and the transcript. Stop the interaction first, then drain, and release the transcript in a finally block. Adds regression tests for output queued during interaction close and for a transcript write failure. Co-Authored-By: Claude Haiku 5.5 Signed-off-by: jay23606 <4602056+jay23606@users.noreply.github.com> --- scripts/nvx_tools/openvmm_process.py | 14 ++++++--- scripts/test_microvm_tests.py | 47 ++++++++++++++++++++++++++++ 2 files changed, 56 insertions(+), 5 deletions(-) diff --git a/scripts/nvx_tools/openvmm_process.py b/scripts/nvx_tools/openvmm_process.py index c7f18cf8c..9459fdb2f 100644 --- a/scripts/nvx_tools/openvmm_process.py +++ b/scripts/nvx_tools/openvmm_process.py @@ -288,11 +288,15 @@ def close(self) -> None: """ if not self._finished and self.process.poll() is None: terminate(self.process) - if not self._log.closed: - self._drain_available() - self._close_log() - self._interaction.close() - self._finished = True + try: + # The reader queues its last chunks while the interaction shuts down, so + # drain only after it has stopped, and release the transcript either way. + self._interaction.close() + if not self._log.closed: + self._drain_available() + finally: + self._close_log() + self._finished = True def _drain_available(self) -> None: """Keep the output still queued; failure paths only, so no scan.""" diff --git a/scripts/test_microvm_tests.py b/scripts/test_microvm_tests.py index 660737812..ac043ae3a 100644 --- a/scripts/test_microvm_tests.py +++ b/scripts/test_microvm_tests.py @@ -4869,6 +4869,53 @@ def test_process_close_writes_output_delivered_after_the_last_wait(self): process.close() self.assertEqual(log_path.read_bytes(), b"LATE\n") + def test_process_close_keeps_output_queued_during_interaction_close(self): + with tempfile.TemporaryDirectory() as temporary: + log_path = Path(temporary) / "output.log" + closed: list[bool] = [] + with ( + patch.object(openvmm_process, "InteractiveProcess") as interaction, + patch.object(openvmm_process.threading, "Thread"), + patch.object(openvmm_process.queue, "Queue") as queues, + ): + interaction.return_value.process.poll.return_value = None + interaction.return_value.close.side_effect = lambda: closed.append(True) + + pending = [b"LAST\n"] + + def get_nowait() -> bytes: + # The reader queues its last chunk only while the interaction shuts down. + if closed and pending: + return pending.pop() + raise queue.Empty + + queues.return_value.get_nowait.side_effect = get_nowait + process = openvmm_process.OpenvmmProcess(["openvmm"], log_path) + process.close() + self.assertEqual(process.output, b"LAST\n") + self.assertEqual(log_path.read_bytes(), b"LAST\n") + + def test_process_close_releases_resources_when_the_transcript_write_fails(self): + with tempfile.TemporaryDirectory() as temporary: + log_path = Path(temporary) / "output.log" + with ( + patch.object(openvmm_process, "InteractiveProcess") as interaction, + patch.object(openvmm_process.threading, "Thread"), + patch.object(openvmm_process.queue, "Queue") as queues, + ): + interaction.return_value.process.poll.return_value = None + queues.return_value.get_nowait.side_effect = [b"LATE\n", queue.Empty] + process = openvmm_process.OpenvmmProcess(["openvmm"], log_path) + failing = MagicMock() + failing.closed = False + failing.write.side_effect = OSError("no space left on device") + process._log.close() + process._log = failing + with self.assertRaises(OSError): + process.close() + interaction.return_value.close.assert_called_once_with() + failing.close.assert_called_once_with() + def test_process_wait_clamps_elapsed_process_timeout(self): with tempfile.TemporaryDirectory() as temporary: with ( From 4440205ee28c73bc3a7fddba4d574f4cdfb077f3 Mon Sep 17 00:00:00 2001 From: jay23606 <4602056+jay23606@users.noreply.github.com> Date: Fri, 16 Oct 2026 16:44:43 -0400 Subject: [PATCH 3/4] microvm tests: drain queued output even when interaction close fails If InteractiveProcess.close raised, the drain was skipped and chunks still queued were missing from both output and the transcript. Drain in a finally around the interaction shutdown, so the transcript is complete whichever step fails. Adds a regression test with queued output and a failing interaction shutdown. Co-Authored-By: Claude Haiku 5.5 Signed-off-by: jay23606 <4602056+jay23606@users.noreply.github.com> --- scripts/nvx_tools/openvmm_process.py | 11 +++++++---- scripts/test_microvm_tests.py | 17 +++++++++++++++++ 2 files changed, 24 insertions(+), 4 deletions(-) diff --git a/scripts/nvx_tools/openvmm_process.py b/scripts/nvx_tools/openvmm_process.py index 9459fdb2f..1c8e0815e 100644 --- a/scripts/nvx_tools/openvmm_process.py +++ b/scripts/nvx_tools/openvmm_process.py @@ -290,10 +290,13 @@ def close(self) -> None: terminate(self.process) try: # The reader queues its last chunks while the interaction shuts down, so - # drain only after it has stopped, and release the transcript either way. - self._interaction.close() - if not self._log.closed: - self._drain_available() + # drain only after it has stopped. Drain even if that shutdown fails, and + # release the transcript either way. + try: + self._interaction.close() + finally: + if not self._log.closed: + self._drain_available() finally: self._close_log() self._finished = True diff --git a/scripts/test_microvm_tests.py b/scripts/test_microvm_tests.py index ac043ae3a..45920f3db 100644 --- a/scripts/test_microvm_tests.py +++ b/scripts/test_microvm_tests.py @@ -4895,6 +4895,23 @@ def get_nowait() -> bytes: self.assertEqual(process.output, b"LAST\n") self.assertEqual(log_path.read_bytes(), b"LAST\n") + def test_process_close_keeps_queued_output_when_interaction_close_fails(self): + with tempfile.TemporaryDirectory() as temporary: + log_path = Path(temporary) / "output.log" + with ( + patch.object(openvmm_process, "InteractiveProcess") as interaction, + patch.object(openvmm_process.threading, "Thread"), + patch.object(openvmm_process.queue, "Queue") as queues, + ): + interaction.return_value.process.poll.return_value = None + interaction.return_value.close.side_effect = OSError("reader stuck") + queues.return_value.get_nowait.side_effect = [b"QUEUED\n", queue.Empty] + process = openvmm_process.OpenvmmProcess(["openvmm"], log_path) + with self.assertRaises(OSError): + process.close() + self.assertEqual(process.output, b"QUEUED\n") + self.assertEqual(log_path.read_bytes(), b"QUEUED\n") + def test_process_close_releases_resources_when_the_transcript_write_fails(self): with tempfile.TemporaryDirectory() as temporary: log_path = Path(temporary) / "output.log" From 2634167ff4d4f0a1b104c71435f7f1b1c73a06aa Mon Sep 17 00:00:00 2001 From: jay23606 <4602056+jay23606@users.noreply.github.com> Date: Mon, 19 Oct 2026 18:22:17 -0400 Subject: [PATCH 4/4] openvmm: keep draining queued output after a transcript write error _drain_available stopped at the first failed transcript write, so chunks still queued behind it never reached process.output. Keep collecting them in memory and re-raise the first error once the queue is empty. Co-Authored-By: Claude Haiku 5.5 Signed-off-by: jay23606 <4602056+jay23606@users.noreply.github.com> --- scripts/nvx_tools/openvmm_process.py | 22 ++++++++++++++++++---- scripts/test_microvm_tests.py | 27 +++++++++++++++++++++++++++ 2 files changed, 45 insertions(+), 4 deletions(-) diff --git a/scripts/nvx_tools/openvmm_process.py b/scripts/nvx_tools/openvmm_process.py index 1c8e0815e..ff4385b2d 100644 --- a/scripts/nvx_tools/openvmm_process.py +++ b/scripts/nvx_tools/openvmm_process.py @@ -302,14 +302,28 @@ def close(self) -> None: self._finished = True def _drain_available(self) -> None: - """Keep the output still queued; failure paths only, so no scan.""" + """Keep the output still queued; failure paths only, so no scan. + + A transcript write error does not stop the drain: every queued chunk is + kept in memory, and the first error is re-raised once the queue is empty. + """ + error: Exception | None = None while True: try: chunk = self._chunks.get_nowait() except queue.Empty: - return - if chunk is not None: - self._record(chunk) + break + if chunk is None: + continue + self._output.extend(chunk) + if error is None: + try: + self._log.write(chunk) + self._log.flush() + except Exception as write_error: + error = write_error + if error is not None: + raise error def _record(self, chunk: bytes) -> None: """Keep a consumed chunk in memory and in the transcript, flushed for readers.""" diff --git a/scripts/test_microvm_tests.py b/scripts/test_microvm_tests.py index 45920f3db..ff6cb3bc8 100644 --- a/scripts/test_microvm_tests.py +++ b/scripts/test_microvm_tests.py @@ -4933,6 +4933,33 @@ def test_process_close_releases_resources_when_the_transcript_write_fails(self): interaction.return_value.close.assert_called_once_with() failing.close.assert_called_once_with() + def test_process_close_keeps_every_queued_chunk_when_the_transcript_write_fails(self): + with tempfile.TemporaryDirectory() as temporary: + log_path = Path(temporary) / "output.log" + with ( + patch.object(openvmm_process, "InteractiveProcess") as interaction, + patch.object(openvmm_process.threading, "Thread"), + patch.object(openvmm_process.queue, "Queue") as queues, + ): + interaction.return_value.process.poll.return_value = None + queues.return_value.get_nowait.side_effect = [ + b"FIRST\n", + b"SECOND\n", + b"THIRD\n", + queue.Empty, + ] + process = openvmm_process.OpenvmmProcess(["openvmm"], log_path) + failing = MagicMock() + failing.closed = False + failing.write.side_effect = OSError("no space left on device") + process._log.close() + process._log = failing + with self.assertRaises(OSError): + process.close() + self.assertEqual(bytes(process._output), b"FIRST\nSECOND\nTHIRD\n") + interaction.return_value.close.assert_called_once_with() + failing.close.assert_called_once_with() + def test_process_wait_clamps_elapsed_process_timeout(self): with tempfile.TemporaryDirectory() as temporary: with (