oprava blokace vlaken

.gitignore

@@ -0,0 +1,4 @@
+app/__pycache__/
+tests/__pycache__/
+__pycache__/
+*.pyc

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:
+        stdout_pump = StreamPump(job_id, "stdout", process.stdout)
-            extend_lock(job)
+        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(
+            returncode = process.wait()
-                    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
 
-            time.sleep(CANCEL_CHECK_SECONDS)
+            for pump in (stdout_pump, stderr_pump):
+                pump.join(STREAM_JOIN_TIMEOUT)
 
-        stdout, stderr = process.communicate()
+                if pump.is_alive():
-        returncode = process.returncode
+                    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:
+                pump.flush()
-            append_job_log(job_id, "stdout", stdout)
 
-        if stderr:
+        if cancelled:
-            append_job_log(job_id, "stderr", stderr)
+            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 ""),
+                    "stdout_length": stdout_pump.length,
-                    "stderr_length": len(stderr or ""),
+                    "stderr_length": stderr_pump.length,
                 },
             )
             update_alert_event_from_job(job, "success", None)

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)