From 9cc00f50fa66d97da9de7f61611d0e7ea99bbe73 Mon Sep 17 00:00:00 2001 From: Balamurali M Date: Wed, 30 Sep 2026 13:11:54 +0530 Subject: [PATCH 1/4] fix(base): Publish command output at most every two seconds parse_output sent the whole output so far to Redis on every line and every progress-bar update. Redis logs each write to its AOF in full, so the AOF grows with the square of the output. On f1 the AOF is 384 MB for 1 MB of data. On a server with the agent on a small root disk, this filled the disk and Redis refused writes with MISCONF. Press polls job output every five seconds (poll_pending_jobs), so two seconds keeps the output it shows fresh. The final output is still published after the loop and by log(). Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01VTnJ7zBU5AidQfFREeTExQ --- agent/base.py | 24 ++++++++++++++++++--- agent/tests/test_base.py | 46 ++++++++++++++++++++++++++++++++++++++++ 2 files changed, 67 insertions(+), 3 deletions(-) create mode 100644 agent/tests/test_base.py diff --git a/agent/base.py b/agent/base.py index 574a51b7f..e263d96f6 100644 --- a/agent/base.py +++ b/agent/base.py @@ -5,6 +5,7 @@ import shutil import subprocess import tempfile +import time import traceback from contextlib import suppress from datetime import datetime @@ -19,10 +20,25 @@ from agent.utils import get_execution_result if TYPE_CHECKING: - from typing import Any + from typing import Any, Callable from agent.job import Job, Step +PUBLISH_INTERVAL_SECONDS = 2 + + +def throttle(function: Callable, seconds: float) -> Callable: + last_call = -seconds + + def throttled(*args): + nonlocal last_call + if time.monotonic() - last_call < seconds: + return + last_call = time.monotonic() + function(*args) + + return throttled + class Base: if TYPE_CHECKING: @@ -135,6 +151,8 @@ def parse_output(self, process) -> str: line = b"" lines = [] prev_char = None + # Each publish rewrites the whole output in Redis and its AOF + publish_lines = throttle(self.publish_lines, PUBLISH_INTERVAL_SECONDS) # This is equivalent of remove_crs # Make sure output matches what'll be shown in the terminal # This won't work for top, htop etc, but good enough to handle progress bars @@ -147,9 +165,9 @@ def parse_output(self, process) -> str: if char == b"\n": lines.append(line.decode(errors="replace")) line = b"" - self.publish_lines(lines) + publish_lines(lines) elif prev_char == b"\r": - self.publish_lines([*lines, line.decode(errors="replace")]) + publish_lines([*lines, line.decode(errors="replace")]) line = b"" line += char else: diff --git a/agent/tests/test_base.py b/agent/tests/test_base.py new file mode 100644 index 000000000..7ed32b0df --- /dev/null +++ b/agent/tests/test_base.py @@ -0,0 +1,46 @@ +from __future__ import annotations + +import io +import unittest +from unittest.mock import MagicMock, patch + +from agent.base import Base + + +class FinishedProcess: + def __init__(self, output: bytes): + self.stdout = io.BytesIO(output) + + def poll(self): + return 0 + + +@patch("agent.base.time.monotonic", return_value=1000.0) +class TestParseOutput(unittest.TestCase): + def setUp(self): + self.base = Base() + self.base.update_redis = MagicMock() + + def test_lines_within_two_seconds_are_published_once_plus_the_final_output(self, _): + output = b"".join(f"line {i}\n".encode() for i in range(1000)) + + self.base.parse_output(FinishedProcess(output)) + + self.assertEqual(self.base.update_redis.call_count, 2) + self.assertEqual(self.base.data["output"], "\n".join(f"line {i}" for i in range(1000))) + + def test_progress_bar_updates_within_two_seconds_are_published_once_plus_the_final_output(self, _): + output = b"".join(f"\r{percent}%".encode() for percent in range(101)) + b"\n" + + self.base.parse_output(FinishedProcess(output)) + + self.assertEqual(self.base.update_redis.call_count, 2) + self.assertEqual(self.base.data["output"], "100%") + + def test_output_is_published_again_after_two_seconds(self, monotonic): + monotonic.side_effect = [1000.0, 1000.0, 1001.0, 1002.0, 1002.0] + + self.base.parse_output(FinishedProcess(b"a\nb\nc\n")) + + # a (first), b skipped at +1s, c at +2s, then the final output + self.assertEqual(self.base.update_redis.call_count, 3) From f16086bfeef9be43f9be1142f1630b1e7bc1c611 Mon Sep 17 00:00:00 2001 From: Balamurali M Date: Wed, 30 Sep 2026 15:16:19 +0530 Subject: [PATCH 2/4] refactor(base): Make the output throttle a class RQ pickles the object that a job method is bound to. A closure cannot be pickled, so an object that stores the throttle as an attribute could not be enqueued. The image builders need this in the next commit. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01VTnJ7zBU5AidQfFREeTExQ --- agent/base.py | 22 ++++++++++++---------- agent/tests/test_base.py | 13 ++++++++++++- 2 files changed, 24 insertions(+), 11 deletions(-) diff --git a/agent/base.py b/agent/base.py index e263d96f6..34faad119 100644 --- a/agent/base.py +++ b/agent/base.py @@ -27,17 +27,19 @@ PUBLISH_INTERVAL_SECONDS = 2 -def throttle(function: Callable, seconds: float) -> Callable: - last_call = -seconds +class Throttle: + """Calls function at most once every `seconds`. A class, not a closure, so RQ can pickle its owner.""" - def throttled(*args): - nonlocal last_call - if time.monotonic() - last_call < seconds: - return - last_call = time.monotonic() - function(*args) + def __init__(self, function: Callable, seconds: float): + self.function = function + self.seconds = seconds + self.last_call = -seconds - return throttled + def __call__(self, *args): + if time.monotonic() - self.last_call < self.seconds: + return + self.last_call = time.monotonic() + self.function(*args) class Base: @@ -152,7 +154,7 @@ def parse_output(self, process) -> str: lines = [] prev_char = None # Each publish rewrites the whole output in Redis and its AOF - publish_lines = throttle(self.publish_lines, PUBLISH_INTERVAL_SECONDS) + publish_lines = Throttle(self.publish_lines, PUBLISH_INTERVAL_SECONDS) # This is equivalent of remove_crs # Make sure output matches what'll be shown in the terminal # This won't work for top, htop etc, but good enough to handle progress bars diff --git a/agent/tests/test_base.py b/agent/tests/test_base.py index 7ed32b0df..351f34570 100644 --- a/agent/tests/test_base.py +++ b/agent/tests/test_base.py @@ -1,10 +1,11 @@ from __future__ import annotations import io +import pickle import unittest from unittest.mock import MagicMock, patch -from agent.base import Base +from agent.base import PUBLISH_INTERVAL_SECONDS, Base, Throttle class FinishedProcess: @@ -44,3 +45,13 @@ def test_output_is_published_again_after_two_seconds(self, monotonic): # a (first), b skipped at +1s, c at +2s, then the final output self.assertEqual(self.base.update_redis.call_count, 3) + + +class TestThrottle(unittest.TestCase): + def test_object_holding_a_throttle_can_be_pickled_for_rq(self): + base = Base() + base.publish_output = Throttle(base.publish_data, PUBLISH_INTERVAL_SECONDS) + + restored = pickle.loads(pickle.dumps(base)) + + self.assertEqual(restored.publish_output.seconds, PUBLISH_INTERVAL_SECONDS) From 6bc82a35f2e7016a58bb6bb4eaf6924488fd2164 Mon Sep 17 00:00:00 2001 From: Balamurali M Date: Wed, 30 Sep 2026 15:17:07 +0530 Subject: [PATCH 3/4] refactor(builder): Use the output throttle from base ImageBuilder and PatchImageBuilder each had a copy of the same throttle. Both now use Throttle from base. Their output is published every two seconds instead of every second, on the monotonic clock instead of the wall clock, and the first update goes out at once. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01VTnJ7zBU5AidQfFREeTExQ --- agent/builder.py | 51 ++++++++++++------------------------------------ 1 file changed, 13 insertions(+), 38 deletions(-) diff --git a/agent/builder.py b/agent/builder.py index 3ddca8ef8..e254f7715 100644 --- a/agent/builder.py +++ b/agent/builder.py @@ -10,7 +10,6 @@ import time import warnings from dataclasses import dataclass, field -from datetime import datetime from pathlib import Path from subprocess import Popen from typing import TYPE_CHECKING, Dict, List, TypedDict @@ -19,7 +18,7 @@ import jinja2 import semantic_version as sv -from agent.base import Base +from agent.base import PUBLISH_INTERVAL_SECONDS, Base, Throttle from agent.build_utils.validations import check_python_syntax, get_package_manager_files from agent.exceptions import AgentException, RegistryDownException from agent.job import Job, Step, job, step @@ -563,7 +562,7 @@ def __init__( self.no_cache = no_cache self.no_push = no_push - self.last_published = datetime.now() + self.publish_output = Throttle(self.publish_data, PUBLISH_INTERVAL_SECONDS) self.build_failed = False self.build_token = build_token self.secret_path = None @@ -657,8 +656,8 @@ def _get_build_environment(self) -> dict: def _publish_docker_build_output(self, result): for line in result: self.output["build"].append(line) - self._publish_throttled_output(False) - self._publish_throttled_output(True) + self.publish_output(self.output) + self.publish_data(self.output) def _wait_for_registry_recovery(self): """Wait for registry to recover after restart""" @@ -689,13 +688,13 @@ def _push_docker_image(self): except RegistryDownException as e: if attempt == max_retries - 1: - self._publish_throttled_output(True) + self.publish_data(self.output) raise Exception("Failed to push image after multiple attempts") from e self._wait_for_registry_recovery() except Exception: - self._publish_throttled_output(True) + self.publish_data(self.output) raise return None @@ -714,19 +713,7 @@ def _push_image(self, client): auth_config=auth_config, ): self.output["push"].append(line) - self._publish_throttled_output(False) - - def _publish_throttled_output(self, flush: bool): - if flush: - self.publish_data(self.output) - return - - now = datetime.now() - if (now - self.last_published).total_seconds() <= 1: - return - - self.last_published = now - self.publish_data(self.output) + self.publish_output(self.output) def _get_image_name(self): return f"{self.image_repository}:{self.image_tag}" @@ -753,7 +740,7 @@ def _run( input_file.close() return_code = process.wait() - self._publish_throttled_output(True) + self.publish_data(self.output) self.build_failed = return_code != 0 self.data.update({"build_failed": self.build_failed}) @@ -793,7 +780,7 @@ def __init__( self.patch_build_app_instructions = patch_build_app_instructions self.container_name = f"patch-build-{build_name}" self.output: Output = {"build": [], "push": []} - self.last_published = datetime.now() + self.publish_output = Throttle(self.publish_data, PUBLISH_INTERVAL_SECONDS) def _get_image_name(self) -> str: return f"{self.image_repository}:{self.image_tag}" @@ -826,7 +813,7 @@ def _start_base_container(self): def _pull_app_updates(self): for patch_build_app_info in self.patch_build_app_instructions: self._pull_app(patch_build_app_info) - self._publish_throttled_output(True) + self.publish_data(self.output) return self.output["build"] def _pull_app(self, patch_build_app_info: PatchBuildAppInfo): @@ -875,21 +862,9 @@ def _docker_exec(self, command: str, publish: bool = True) -> str: output = result.get("output", "") if isinstance(result, dict) else "" if publish: self.output["build"].append(output) - self._publish_throttled_output(False) + self.publish_output(self.output) return output - def _publish_throttled_output(self, flush: bool) -> None: - if flush: - self.publish_data(self.output) - return - - now = datetime.now() - if (now - self.last_published).total_seconds() <= 1: - return - - self.last_published = now - self.publish_data(self.output) - @step("Commit Image") def _commit_patch_image(self): self.execute( @@ -913,9 +888,9 @@ def _push_patch_image(self): auth_config=auth_config, ): self.output["push"].append(line) - self._publish_throttled_output(False) + self.publish_output(self.output) - self._publish_throttled_output(True) + self.publish_data(self.output) return self.output["push"] def _cleanup_container(self): From 5a5919dc8a0dd09dcf474e5e6dc8d3e27b72141b Mon Sep 17 00:00:00 2001 From: Balamurali M Date: Mon, 5 Oct 2026 13:35:04 +0530 Subject: [PATCH 4/4] docs(base): Note that Throttle drops calls between intervals Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_017JQShbx4JgSX4qHARVx3ty --- agent/base.py | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/agent/base.py b/agent/base.py index 34faad119..9f4749e70 100644 --- a/agent/base.py +++ b/agent/base.py @@ -28,7 +28,10 @@ class Throttle: - """Calls function at most once every `seconds`. A class, not a closure, so RQ can pickle its owner.""" + """Calls function at most once every `seconds`, dropping calls made in between. + + A class, not a closure, so RQ can pickle its owner. + """ def __init__(self, function: Callable, seconds: float): self.function = function