diff --git a/src/ci_orchestrator/transport/winrm.py b/src/ci_orchestrator/transport/winrm.py index 5ef4ff0..d9fa6f1 100644 --- a/src/ci_orchestrator/transport/winrm.py +++ b/src/ci_orchestrator/transport/winrm.py @@ -51,11 +51,23 @@ def _info_text(record: Any) -> str: return str(data) if data is not None else str(record) -def _drain(seq: list[Any], start: int, emit: Callable[[str], None]) -> int: - """Emit ``str()`` of every item in ``seq[start:]``; return the new length.""" - for item in seq[start:]: - emit(str(item)) - return len(seq) +def _drain_clear( + seq: list[Any], render: Callable[[Any], str], emit: Callable[[str], None] +) -> None: + """Emit ``render(item)`` for each item in ``seq`` then drop those items. + + Clearing the consumed prefix is essential: pypsrp keeps every output/info/ + error record in these lists for the whole invocation, and each is a heavy + deserialised PSObject. A long prepare (Windows Update + toolchain, thousands + of lines) would otherwise grow the host process RAM without bound. We run on + the same thread as ``poll_invoke`` so nothing appends while we mutate. + """ + if not seq: + return + n = len(seq) + for item in seq[:n]: + emit(render(item)) + del seq[:n] class WinRmTransport: @@ -224,20 +236,23 @@ class WinRmTransport: ps = PowerShell(pool) ps.add_script(script) ps.begin_invoke() - out_i = info_i = err_i = 0 running = (PSInvocationState.NOT_STARTED, PSInvocationState.RUNNING) + had_error = False while ps.state in running: ps.poll_invoke() - out_i = _drain(ps.output, out_i, emit_out) - info_i = _drain( - [_info_text(r) for r in ps.streams.information], info_i, on_line - ) - err_i = _drain([str(e) for e in ps.streams.error], err_i, emit_err) + _drain_clear(ps.output, str, emit_out) + _drain_clear(ps.streams.information, _info_text, on_line) + if ps.streams.error: + had_error = True + _drain_clear(ps.streams.error, str, emit_err) time.sleep(0.5) ps.end_invoke() - _drain(ps.output, out_i, emit_out) - _drain([str(e) for e in ps.streams.error], err_i, emit_err) - returncode = 1 if ps.streams.error else 0 + _drain_clear(ps.output, str, emit_out) + _drain_clear(ps.streams.information, _info_text, on_line) + if ps.streams.error: + had_error = True + _drain_clear(ps.streams.error, str, emit_err) + returncode = 1 if had_error else 0 finally: with contextlib.suppress(Exception): pool.close()