From 8320c08827287ec33e0b198df916e478c207d4ff Mon Sep 17 00:00:00 2001 From: JiriUhlir <149317995+JiriUhlir@users.noreply.github.com> Date: Wed, 26 Aug 2026 09:07:31 +0200 Subject: [PATCH] oprava blokace vlaken --- .gitignore | 4 + app/main.py | 154 ++++++++++++++--- tests/test_execute_job_streaming.py | 245 ++++++++++++++++++++++++++++ 3 files changed, 381 insertions(+), 22 deletions(-) create mode 100644 .gitignore create mode 100644 tests/test_execute_job_streaming.py diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..38e736f --- /dev/null +++ b/.gitignore @@ -0,0 +1,4 @@ +app/__pycache__/ +tests/__pycache__/ +__pycache__/ +*.pyc diff --git a/app/main.py b/app/main.py index 4acc9d1..75b8561 100644 --- a/app/main.py +++ b/app/main.py @@ -20,6 +20,8 @@ WORKER_ID = os.getenv("APPFACTORY_WORKER_ID", socket.gethostname()) POLL_SECONDS = int(os.getenv("APPFACTORY_WORKER_POLL_SECONDS", "3")) HEARTBEAT_SECONDS = int(os.getenv("APPFACTORY_WORKER_HEARTBEAT_SECONDS", "10")) CANCEL_CHECK_SECONDS = int(os.getenv("APPFACTORY_CANCEL_CHECK_SECONDS", "2")) +LOG_FLUSH_CHARS = int(os.getenv("APPFACTORY_LOG_FLUSH_CHARS", "8000")) +STREAM_JOIN_TIMEOUT = int(os.getenv("APPFACTORY_STREAM_JOIN_TIMEOUT", "60")) app = FastAPI(title="AppFactory Worker") @@ -474,6 +476,89 @@ def update_alert_event_from_job(job: dict, status: str, error_text: str | None): con.close() +class StreamPump: + """Reads one subprocess stream in a background thread and stores it in the job log. + + The subprocess writes into an OS pipe with a limited buffer. If nobody reads + the pipe while the process runs, the child blocks on write and never exits, + so the worker would wait forever. This pump keeps both pipes drained from + the moment the process starts. + """ + + def __init__(self, job_id: int, stream_name: str, stream): + self.job_id = job_id + self.stream_name = stream_name + self.stream = stream + self.length = 0 + self._buffer = [] + self._buffer_chars = 0 + self._lock = threading.Lock() + self._thread = threading.Thread( + target=self._run, + name=f"job-{job_id}-{stream_name}", + daemon=True, + ) + + def start(self): + self._thread.start() + + def is_alive(self) -> bool: + return self._thread.is_alive() + + def join(self, timeout: float | None = None): + self._thread.join(timeout) + + def flush(self): + # The lock is held across the write so that the reader thread and the + # main loop cannot store two chunks of the same stream out of order. + with self._lock: + if not self._buffer: + return + + message = "".join(self._buffer) + self._buffer = [] + self._buffer_chars = 0 + + try: + append_job_log(self.job_id, self.stream_name, message) + except Exception as exc: + print(f"Job log write error ({self.stream_name}): {exc}", flush=True) + + def _run(self): + try: + for chunk in iter(self.stream.readline, ""): + with self._lock: + self.length += len(chunk) + self._buffer.append(chunk) + self._buffer_chars += len(chunk) + should_flush = self._buffer_chars >= LOG_FLUSH_CHARS + + if should_flush: + self.flush() + + except Exception as exc: + print(f"Stream reader error ({self.stream_name}): {exc}", flush=True) + try: + append_job_log( + self.job_id, + "system", + f"Error while reading {self.stream_name}: {exc}", + ) + except Exception: + pass + + finally: + try: + self.flush() + except Exception: + pass + + try: + self.stream.close() + except Exception: + pass + + def terminate_process_group(process: subprocess.Popen, job_id: int): append_job_log(job_id, "system", "Cancel requested. Terminating process group...") @@ -521,34 +606,59 @@ def execute_job(job: dict): start_new_session=True, ) - while process.poll() is None: - extend_lock(job) + stdout_pump = StreamPump(job_id, "stdout", process.stdout) + stderr_pump = StreamPump(job_id, "stderr", process.stderr) - status = get_job_status(job_id) + stdout_pump.start() + stderr_pump.start() - if status == "cancelled_requested": + cancelled = False + + try: + while process.poll() is None: + extend_lock(job) + + stdout_pump.flush() + stderr_pump.flush() + + status = get_job_status(job_id) + + if status == "cancelled_requested": + terminate_process_group(process, job_id) + cancelled = True + break + + time.sleep(CANCEL_CHECK_SECONDS) + + finally: + if process.poll() is None: terminate_process_group(process, job_id) - update_job_status( - job_id, - "cancelled", - error_text="Job cancelled by user request", - result={"cancelled": True}, - ) - update_alert_event_from_job(job, "cancelled", "Job cancelled by user request") - append_job_log(job_id, "system", "Job cancelled.") - return + returncode = process.wait() - time.sleep(CANCEL_CHECK_SECONDS) + for pump in (stdout_pump, stderr_pump): + pump.join(STREAM_JOIN_TIMEOUT) - stdout, stderr = process.communicate() - returncode = process.returncode + if pump.is_alive(): + append_job_log( + job_id, + "system", + f"Stream {pump.stream_name} did not reach EOF within {STREAM_JOIN_TIMEOUT} s." + " Some output may be missing from the log.", + ) - if stdout: - append_job_log(job_id, "stdout", stdout) + pump.flush() - if stderr: - append_job_log(job_id, "stderr", stderr) + if cancelled: + update_job_status( + job_id, + "cancelled", + error_text="Job cancelled by user request", + result={"cancelled": True}, + ) + update_alert_event_from_job(job, "cancelled", "Job cancelled by user request") + append_job_log(job_id, "system", "Job cancelled.") + return if get_job_status(job_id) == "cancelled_requested": update_job_status( @@ -567,8 +677,8 @@ def execute_job(job: dict): "success", result={ "returncode": returncode, - "stdout_length": len(stdout or ""), - "stderr_length": len(stderr or ""), + "stdout_length": stdout_pump.length, + "stderr_length": stderr_pump.length, }, ) update_alert_event_from_job(job, "success", None) diff --git a/tests/test_execute_job_streaming.py b/tests/test_execute_job_streaming.py new file mode 100644 index 0000000..5b540d5 --- /dev/null +++ b/tests/test_execute_job_streaming.py @@ -0,0 +1,245 @@ +"""Regression tests for job output streaming in execute_job(). + +The subprocess writes into OS pipes with a limited buffer. If the worker does +not read stdout and stderr while the process runs, the child blocks on write +and the job never finishes. These tests run a child that produces far more +output than any pipe buffer can hold. +""" + +import json +import os +import sqlite3 +import sys +import tempfile +import threading +import unittest +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).resolve().parents[1])) + +from app import main # noqa: E402 + +# Much larger than the pipe buffer on Linux (64 KiB) and Windows (4 KiB). +CHUNK_LINES = 4000 +LINE = "x" * 256 + +CHILD_SCRIPT = r""" +import sys + +line = "x" * 256 + +for i in range({lines}): + sys.stdout.write("out %d %s\n" % (i, line)) + sys.stderr.write("err %d %s\n" % (i, line)) + +sys.stdout.flush() +sys.stderr.flush() +sys.exit({exit_code}) +""" + +SCHEMA = """ +CREATE TABLE jobs ( + id INTEGER PRIMARY KEY, + type TEXT NOT NULL, + target_type TEXT, + target_id TEXT, + status TEXT NOT NULL, + payload_json TEXT, + source TEXT, + created_by_user_id INTEGER, + created_by_username TEXT, + created_by_display_name TEXT, + worker_id TEXT, + started_at TEXT, + finished_at TEXT, + error_text TEXT, + result_json TEXT +); + +CREATE TABLE job_logs ( + id INTEGER PRIMARY KEY, + job_id INTEGER NOT NULL, + stream TEXT NOT NULL, + message TEXT, + created_at TEXT +); + +CREATE TABLE deployment_locks ( + target_type TEXT NOT NULL, + target_id TEXT NOT NULL, + job_id INTEGER, + worker_id TEXT, + acquired_at TEXT, + expires_at TEXT, + PRIMARY KEY (target_type, target_id) +); + +CREATE TABLE workers ( + id TEXT PRIMARY KEY, + status TEXT, + started_at TEXT, + last_seen_at TEXT, + current_job_id INTEGER, + metadata_json TEXT +); + +CREATE TABLE alert_events ( + id INTEGER PRIMARY KEY, + status TEXT, + error_text TEXT, + processed_at TEXT +); +""" + + +class ExecuteJobStreamingTest(unittest.TestCase): + def setUp(self): + self._tmp_dir = tempfile.TemporaryDirectory() + self.addCleanup(self._tmp_dir.cleanup) + + db_file = Path(self._tmp_dir.name) / "appfactory.db" + + con = sqlite3.connect(db_file) + con.executescript(SCHEMA) + con.commit() + con.close() + + self._original_db_file = main.DB_FILE + self._original_command_for_job = main.command_for_job + self._original_cancel_check = main.CANCEL_CHECK_SECONDS + + main.DB_FILE = db_file + main.CANCEL_CHECK_SECONDS = 1 + + self.addCleanup(self._restore) + + self.db_file = db_file + + def _restore(self): + main.DB_FILE = self._original_db_file + main.command_for_job = self._original_command_for_job + main.CANCEL_CHECK_SECONDS = self._original_cancel_check + + def _connect(self): + con = sqlite3.connect(self.db_file) + con.row_factory = sqlite3.Row + return con + + def _insert_job(self, job_type: str, target_id: str) -> dict: + con = self._connect() + cur = con.execute( + """ + INSERT INTO jobs (type, target_type, target_id, status, payload_json, source) + VALUES (?, ?, ?, 'running', '{}', 'test') + """, + (job_type, job_type, target_id), + ) + job_id = cur.lastrowid + con.commit() + + row = con.execute("SELECT * FROM jobs WHERE id = ?", (job_id,)).fetchone() + con.close() + + return dict(row) + + def _use_child(self, exit_code: int): + script = CHILD_SCRIPT.format(lines=CHUNK_LINES, exit_code=exit_code) + script_file = Path(self._tmp_dir.name) / f"child_{exit_code}.py" + script_file.write_text(script, encoding="utf-8") + + main.command_for_job = lambda job: [sys.executable, str(script_file)] + + def _run_job(self, job: dict, timeout: float = 120.0): + error = {} + + def runner(): + try: + main.execute_job(job) + except BaseException as exc: # pragma: no cover - reported below + error["exc"] = exc + + thread = threading.Thread(target=runner, daemon=True) + thread.start() + thread.join(timeout) + + self.assertFalse( + thread.is_alive(), + f"execute_job did not finish within {timeout} s (deadlock on a full pipe buffer)", + ) + + if "exc" in error: + raise error["exc"] + + def _job_row(self, job_id: int) -> sqlite3.Row: + con = self._connect() + row = con.execute("SELECT * FROM jobs WHERE id = ?", (job_id,)).fetchone() + con.close() + return row + + def _log_text(self, job_id: int, stream: str) -> str: + con = self._connect() + rows = con.execute( + "SELECT message FROM job_logs WHERE job_id = ? AND stream = ? ORDER BY id ASC", + (job_id, stream), + ).fetchall() + con.close() + return "".join(row["message"] for row in rows) + + def test_large_output_on_both_streams_finishes(self): + """stdout and stderr are drained while the process runs, so it can exit.""" + self._use_child(exit_code=0) + + job = self._insert_job("deploy_app", "demo-app") + self._run_job(job) + + row = self._job_row(job["id"]) + self.assertEqual(row["status"], "success") + + result = json.loads(row["result_json"]) + self.assertEqual(result["returncode"], 0) + + expected_length = CHUNK_LINES * (len(LINE) + len("out 0 \n")) + self.assertGreater(result["stdout_length"], 64 * 1024) + self.assertGreater(result["stderr_length"], 64 * 1024) + + stdout_text = self._log_text(job["id"], "stdout") + stderr_text = self._log_text(job["id"], "stderr") + + self.assertEqual(len(stdout_text), result["stdout_length"]) + self.assertEqual(len(stderr_text), result["stderr_length"]) + self.assertGreaterEqual(len(stdout_text), expected_length - CHUNK_LINES * 4) + + # Nothing may be lost and nothing stored twice. + self.assertEqual(stdout_text.count("\nout "), CHUNK_LINES - 1) + self.assertEqual(stderr_text.count("\nerr "), CHUNK_LINES - 1) + + self.assertTrue(stdout_text.startswith("out 0 ")) + self.assertIn(f"out {CHUNK_LINES - 1} ", stdout_text) + self.assertIn(f"err {CHUNK_LINES - 1} ", stderr_text) + + def test_large_output_with_failure_is_marked_failed(self): + """The return code survives the streaming rewrite.""" + self._use_child(exit_code=3) + + job = self._insert_job("run_script", "cleanup.sh") + self._run_job(job) + + row = self._job_row(job["id"]) + self.assertEqual(row["status"], "failed") + self.assertIn("3", row["error_text"]) + + result = json.loads(row["result_json"]) + self.assertEqual(result["returncode"], 3) + + self.assertEqual( + self._log_text(job["id"], "stdout").count("\nout "), + CHUNK_LINES - 1, + ) + self.assertEqual( + self._log_text(job["id"], "stderr").count("\nerr "), + CHUNK_LINES - 1, + ) + + +if __name__ == "__main__": + unittest.main(verbosity=2)