From 2321386113eed0066dd7c8694c33a5c9f3c57b49 Mon Sep 17 00:00:00 2001 From: Robert Ursu Date: Tue, 4 Aug 2026 15:41:10 +0300 Subject: [PATCH] fix(cli): bound how much log a single drain reads into memory MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The tailer did f.read() with no limit, so one drain pulled everything written since the last poll — and the decoded str costs more again. Normally the 250 ms poll keeps that tiny, but a job that logs heavily between polls, a starved tailer, or the final drain after a long run could all pull a large log in at once. The same read-it-all-to-forward-it shape has OOMed pods on the handler side. Each pass now reads at most LOG_READ_CHUNK_BYTES and loops until caught up, so peak memory is the window rather than the backlog. Two cases only reachable once the read is bounded: - A chunk boundary landing mid-line: trimmed back to the last newline, which is also what keeps the decode safe, since \n never appears inside a multi-byte UTF-8 sequence. - A single line longer than the window: emitted as a fragment rather than held. Buffering it without bound would defeat the point, and stalling on it would wedge the tailer permanently. Co-Authored-By: Claude Opus 5 --- .../uipath/src/uipath/_cli/_server_jobs.py | 61 +++++++++++++------ packages/uipath/tests/cli/test_server_logs.py | 49 +++++++++++++++ 2 files changed, 90 insertions(+), 20 deletions(-) diff --git a/packages/uipath/src/uipath/_cli/_server_jobs.py b/packages/uipath/src/uipath/_cli/_server_jobs.py index a984c3886..a9662974a 100644 --- a/packages/uipath/src/uipath/_cli/_server_jobs.py +++ b/packages/uipath/src/uipath/_cli/_server_jobs.py @@ -189,6 +189,10 @@ def build_result_payload(job_key: str, outcome: dict[str, Any]) -> dict[str, Any LOG_POLL_SECONDS = 0.25 LOG_BATCH_MAX_LINES = 200 +# Cap on how much log a single drain pulls into memory. Without it a job that logs +# heavily between polls -- or the final drain after a long run -- reads the whole +# remainder at once, and the decoded str costs more again. +LOG_READ_CHUNK_BYTES = 256 * 1024 LOG_FLUSH_TIMEOUT_SECONDS = 10 # How long a cancelled job gets to unwind before we admit the stop did not take. @@ -249,31 +253,48 @@ def stop(self) -> None: self._stop.set() async def _drain(self, final: bool = False) -> None: - try: - if not os.path.exists(self.path): + """Forward whatever is new, a bounded window at a time until caught up.""" + while True: + try: + if not os.path.exists(self.path): + return + # Binary with explicit offsets: a text-mode tell() cookie is opaque, and + # we need to rewind past an unterminated tail. + with open(self.path, "rb") as f: + f.seek(self._offset) + chunk = f.read(LOG_READ_CHUNK_BYTES) + except OSError: return - # Binary with explicit offsets: a text-mode tell() cookie is opaque, and we - # need to rewind past an unterminated tail. - with open(self.path, "rb") as f: - f.seek(self._offset) - chunk = f.read() - except OSError: - return - if not chunk: - return - - # A logging handler writes the record and only then flushes, so the tail can be - # half a line. Leave it for the next poll rather than reporting a fragment as a - # complete entry — except on the final drain, where nothing more is coming. - if not final and not chunk.endswith(b"\n"): - cut = chunk.rfind(b"\n") - if cut == -1: + if not chunk: return - chunk = chunk[: cut + 1] - self._offset += len(chunk) + # A short read means we reached the end of the file; a full one means there + # is more waiting and we should come back round. + more_pending = len(chunk) == LOG_READ_CHUNK_BYTES + + if not chunk.endswith(b"\n"): + # A logging handler writes the record and only then flushes, so the tail + # can be half a line. + cut = chunk.rfind(b"\n") + if cut != -1: + # Trim to the last complete line. Safe to decode: \n never appears + # inside a multi-byte UTF-8 sequence. + chunk = chunk[: cut + 1] + elif not (more_pending or final): + # Partial line still being written — wait for its newline. + return + # Otherwise: a single line longer than the read window, or the last drain + # with nothing more coming. Emit the fragment rather than buffering + # without bound. + + self._offset += len(chunk) + await self._emit(chunk) + + if not more_pending: + return + async def _emit(self, chunk: bytes) -> None: batch: list[dict[str, Any]] = [] for raw in chunk.decode("utf-8", errors="replace").splitlines(): line = raw.rstrip("\r") diff --git a/packages/uipath/tests/cli/test_server_logs.py b/packages/uipath/tests/cli/test_server_logs.py index 3b8094765..69ec80625 100644 --- a/packages/uipath/tests/cli/test_server_logs.py +++ b/packages/uipath/tests/cli/test_server_logs.py @@ -294,3 +294,52 @@ async def test_tailer_tolerates_a_missing_file(tmp_path): await asyncio.wait_for(tailer.run(), timeout=5) assert callback.batches == [] + + +# --------------------------------------------------------------------------- # +# bounded reads # +# --------------------------------------------------------------------------- # + + +async def test_tailer_catches_up_across_multiple_read_windows(tmp_path, monkeypatch): + """A job that logs more than one window between polls must still be forwarded in + full — bounding the read must not mean dropping the remainder.""" + import uipath._cli._server_jobs as jobs + + monkeypatch.setattr(jobs, "LOG_READ_CHUNK_BYTES", 512) + + log_file = tmp_path / "execution.log" + total = 200 # ~40 bytes each, comfortably several windows + log_file.write_bytes( + b"".join(b"[2026-07-29 17:00:00,000][INFO] line-%d\n" % i for i in range(total)) + ) + + callback = RecordingCallback() + tailer = JobLogTailer("job-1", str(log_file), callback) + tailer.stop() + await asyncio.wait_for(tailer.run(), timeout=10) + + messages = [line["message"] for line in callback.lines] + assert len(messages) == total + assert messages[0] == "line-0" + assert messages[-1] == f"line-{total - 1}" + + +async def test_tailer_emits_a_line_longer_than_the_read_window(tmp_path, monkeypatch): + """A single pathologically long line must not stall the tailer forever: bounded + memory beats perfect line integrity here.""" + import uipath._cli._server_jobs as jobs + + monkeypatch.setattr(jobs, "LOG_READ_CHUNK_BYTES", 64) + + log_file = tmp_path / "execution.log" + log_file.write_bytes(b"x" * 500 + b"\n[2026-07-29 17:00:00,000][INFO] after\n") + + callback = RecordingCallback() + tailer = JobLogTailer("job-1", str(log_file), callback) + tailer.stop() + await asyncio.wait_for(tailer.run(), timeout=10) + + joined = "".join(line["message"] for line in callback.lines) + assert "x" * 500 in joined, "the long line must be forwarded, even if fragmented" + assert any(line["message"] == "after" for line in callback.lines)