fix(ansible): runs end cleanly: output read in chunks, the process group always ended, a cancelled task recorded (#5156)

- Output is read in 64 KiB chunks and split into lines here; a line past
  64 KiB is cut and marked instead of raising out of readline's limit.
- ansible-playbook runs in its own session; whatever ends the read (finish,
  an error, the task being cancelled) SIGTERMs then SIGKILLs its whole group,
  so no ssh or worker it forked outlives the run.
- A run whose task is cancelled (shutdown) records 'cancelled' with
  finished_at before re-raising, instead of staying 'running'.
- A cancel that lands after the process exited 0 no longer turns a
  successful run into 'cancelled'.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

steward/ansible/executor.py

@@ -6,6 +6,7 @@ import logging
 import os
 import re
 import shutil
+import signal
 import tempfile
 import time
 from dataclasses import dataclass
@@ -26,6 +27,8 @@ _FLUSH_SECS = 5.0
 _MEM_LINES_CAP = 5000   # bounded per-run in-memory replay buffer (SSE late-joiners)
 _MEM_TTL_SECS = 300     # drop in-memory state this long after a run completes
 _FAILURE_CAP = 50       # max failed-task lines captured into structured results
+_READ_CHUNK = 65536     # output is read in chunks, never a whole line at once (#5156)
+_MAX_LINE_BYTES = 65536  # a longer line is cut here and marked, not allowed to stall
 
 # Full run logs persist here (survive restart + the 1 MB DB cap). Override via env.
 ARTIFACT_DIR = os.environ.get("STEWARD_ANSIBLE_LOG_DIR", "/data/ansible/runs")
@@ -261,20 +264,63 @@ async def _set_status(app: "Quart", run_id: str, status) -> None:
             )
 
 
-async def _terminate_then_kill(proc: asyncio.subprocess.Process, grace: float = 10.0) -> None:
+def _signal_group(proc: asyncio.subprocess.Process, sig: int) -> None:
+    """Send sig to the run's whole process group (ansible-playbook and the
+    ssh and worker processes it forks), or to the process alone when it has
+    no group of its own."""
     try:
-        proc.terminate()
+        os.killpg(proc.pid, sig)
     except ProcessLookupError:
-        return
+        pass
-    try:
+    except OSError:
-        await asyncio.wait_for(proc.wait(), timeout=grace)
-    except asyncio.TimeoutError:
         try:
-            proc.kill()
+            proc.send_signal(sig)
         except ProcessLookupError:
             pass
 
 
+async def _terminate_then_kill(proc: asyncio.subprocess.Process, grace: float = 10.0) -> None:
+    """SIGTERM the run's process group, then SIGKILL it if it outlives grace.
+    The group is killed even once the leader has exited, so no worker it
+    forked is left behind (#5156)."""
+    if proc.returncode is None:
+        _signal_group(proc, signal.SIGTERM)
+        try:
+            await asyncio.wait_for(proc.wait(), timeout=grace)
+        except asyncio.TimeoutError:
+            pass
+    _signal_group(proc, signal.SIGKILL)
+
+
+async def _lines(stream: asyncio.StreamReader):
+    """The stream's lines, read in chunks so no line length can stall or break
+    the reader: a line past _MAX_LINE_BYTES is cut there and marked, and the
+    rest of it is dropped up to its newline (#5156)."""
+    pending = b""
+    skipping = False
+    while True:
+        chunk = await stream.read(_READ_CHUNK)
+        if not chunk:
+            break
+        pending += chunk
+        while True:
+            nl = pending.find(b"\n")
+            if nl < 0:
+                break
+            line, pending = pending[:nl], pending[nl + 1:]
+            if skipping:
+                skipping = False
+                continue
+            yield line
+        if len(pending) > _MAX_LINE_BYTES and not skipping:
+            yield pending[:_MAX_LINE_BYTES] + b" [line cut at %d bytes]" % _MAX_LINE_BYTES
+            pending, skipping = b"", True
+        elif skipping:
+            pending = b""
+    if pending and not skipping:
+        yield pending
+
+
 def cancel_run(run_id: str) -> bool:
     """Request cancellation of a run. Terminates the process if it's running;
     a queued run is flagged so start_run aborts before launch. Returns False
@@ -404,6 +450,8 @@ async def start_run(
 
     final_status = "failed"
     logf = None
+    proc = None
+    interrupted = False
     try:
         if run_id in _cancelled:  # cancelled while queued — abort before launching
             raise _Aborted
@@ -463,17 +511,19 @@ async def start_run(
                 if ssh_user:
                     user_args += ["--user", ssh_user]
 
+            # Its own session, so the whole group (ssh, forks) can be signalled.
             proc = await asyncio.create_subprocess_exec(
                 *cmd, *cred_args, *boot_args, *user_args, *ev_args,
                 stdout=asyncio.subprocess.PIPE,
                 stderr=asyncio.subprocess.STDOUT,
                 cwd=cwd,
                 env=ansible_env(creds, os.environ),
+                start_new_session=True,
             )
             _run_procs[run_id] = proc
             assert proc.stdout is not None
 
-            async for raw_line in proc.stdout:
+            async for raw_line in _lines(proc.stdout):
                 line = scrub(raw_line.decode(errors="replace").rstrip("\n\r"), secrets)
                 _broadcast(run_id, line)
                 if logf is not None:
@@ -496,7 +546,8 @@ async def start_run(
 
             await proc.wait()
             recap = parse_recap(_run_lines.get(run_id, []))
-            if run_id in _cancelled:
+            # A cancel that lands after the run finished doesn't undo its result.
+            if run_id in _cancelled and proc.returncode != 0:
                 final_status = "cancelled"
             elif proc.returncode != 0:
                 final_status = "failed"
@@ -511,10 +562,21 @@ async def start_run(
             else:
                 final_status = "success"
         finally:
+            # Whatever ended the read (the run finishing, an error, the task
+            # being cancelled), nothing the run started outlives it.
+            if proc is not None:
+                await asyncio.shield(_terminate_then_kill(proc, grace=5.0))
             shutil.rmtree(tmpdir, ignore_errors=True)
 
     except _Aborted:
         final_status = "cancelled"
+    except asyncio.CancelledError:
+        # The task itself was cancelled (shutdown): record that the run did
+        # not finish rather than leave it marked running (#5156).
+        final_status, interrupted = "cancelled", True
+        msg = "[run error] Steward stopped the run before it finished."
+        _broadcast(run_id, msg)
+        db_output = (db_output + "\n" + msg) if db_output else msg
     except Exception as exc:
         logger.exception("Ansible run %s failed with exception", run_id)
         final_status = "cancelled" if run_id in _cancelled else "failed"
@@ -533,5 +595,11 @@ async def start_run(
         sem.release()
 
     results = {"hosts": parse_recap(_run_lines.get(run_id, [])), "failures": failures}
-    await _flush(final_status, results=results)
+    try:
+        await asyncio.shield(_flush(final_status, results=results))
+    except Exception:
+        logger.exception("Ansible run %s: could not record its final status %s",
+                         run_id, final_status)
     _broadcast(run_id, _Done(status=final_status))
+    if interrupted:
+        raise asyncio.CancelledError

tests/core/test_ansible_executor.py

@@ -28,6 +28,34 @@ def test_parse_recap_empty_without_recap():
     assert parse_recap(["TASK [x] ***", "ok: [h]"]) == {}
 
 
+class _Stream:
+    def __init__(self, chunks):
+        self.chunks = list(chunks)
+
+    async def read(self, _n):
+        return self.chunks.pop(0) if self.chunks else b""
+
+
+def _read_lines(chunks, monkeypatch, limit=8):
+    import asyncio
+    from steward.ansible import executor
+    monkeypatch.setattr(executor, "_MAX_LINE_BYTES", limit)
+
+    async def _go():
+        return [line async for line in executor._lines(_Stream(chunks))]
+    return asyncio.run(_go())
+
+
+def test_lines_split_across_chunks(monkeypatch):
+    assert _read_lines([b"ab", b"c\nde", b"f\n", b"tail"], monkeypatch) == [b"abc", b"def", b"tail"]
+
+
+def test_a_line_past_the_limit_is_cut_and_the_rest_dropped(monkeypatch):
+    # #5156: a long line can't stall or break the reader; the next line is intact.
+    got = _read_lines([b"0123456789", b"abcdef", b"gh\nnext\n"], monkeypatch)
+    assert got == [b"0123456789"[:8] + b" [line cut at 8 bytes]", b"next"]
+
+
 def test_cancel_run_flags_untracked_run():
     rid = "nonexistent-run-id"
     assert cancel_run(rid) is True

tests/integration/test_ansible_run_end.py

@@ -0,0 +1,166 @@
+"""Integration: how an Ansible run ends (#5156).
+
+ansible-playbook is replaced by a small Python script, run as a real process
+through the executor, so reading, the process group and the final status are
+the real paths. Needs Postgres for the run row.
+"""
+from __future__ import annotations
+
+import asyncio
+import os
+import sys
+import time
+import uuid
+
+import pytest
+
+pytestmark = pytest.mark.integration
+
+RECAP = ("PLAY RECAP ****\\n"
+         "localhost : ok=1 changed=0 unreachable=0 failed=0 skipped=0 rescued=0 ignored=0")
+
+# Starts a child in the same process group and records its pid, so a test can
+# check nothing the run started outlives it.
+SPAWN_CHILD = (
+    "import subprocess, sys, time\n"
+    "c = subprocess.Popen([sys.executable, '-c', 'import time; time.sleep(60)'])\n"
+    "open(sys.argv[1], 'w').write(str(c.pid))\n"
+)
+
+
+@pytest.fixture
+def app():
+    if not os.environ.get("STEWARD_DATABASE_URL"):
+        pytest.skip("needs Postgres")
+    from steward.app import create_app
+    return create_app(testing=False)
+
+
+def _use_script(monkeypatch, script: str, *args: str) -> None:
+    from steward.ansible import executor
+    monkeypatch.setattr(executor, "build_ansible_command",
+                        lambda *_a, **_k: [sys.executable, "-c", script, *args])
+
+
+async def _new_run(app) -> str:
+    from steward.models.ansible import AnsibleRun, AnsibleRunStatus
+    run_id = str(uuid.uuid4())
+    async with app.db_sessionmaker() as s:
+        async with s.begin():
+            s.add(AnsibleRun(id=run_id, playbook_path="t.yml", inventory_path="inv.ini",
+                             source_name="t", triggered_by=None,
+                             status=AnsibleRunStatus.running))
+    return run_id
+
+
+async def _row(app, run_id):
+    from sqlalchemy import select
+    from steward.models.ansible import AnsibleRun
+    async with app.db_sessionmaker() as s:
+        return (await s.execute(select(AnsibleRun).where(AnsibleRun.id == run_id))).scalar_one()
+
+
+def _gone(pid: int, within: float = 5.0) -> bool:
+    """The process has exited (a zombie left for init to reap counts)."""
+    deadline = time.monotonic() + within
+    while time.monotonic() < deadline:
+        try:
+            with open(f"/proc/{pid}/status") as f:
+                if "State:\tZ" in f.read():
+                    return True
+        except FileNotFoundError:
+            return True
+        time.sleep(0.1)
+    return False
+
+
+def test_a_line_longer_than_the_reader_limit_is_cut_not_fatal(app, monkeypatch, tmp_path):
+    from steward.ansible import executor
+    from steward.models.ansible import AnsibleRunStatus
+    _use_script(monkeypatch, f"print('x' * 200000); print('after'); print('{RECAP}')")
+
+    async def _go():
+        run_id = await _new_run(app)
+        await executor.start_run(app, run_id, "t.yml", "inv.ini", str(tmp_path))
+        return await _row(app, run_id)
+
+    run = asyncio.run(_go())
+    assert run.status == AnsibleRunStatus.success, run.output
+    assert f"[line cut at {executor._MAX_LINE_BYTES} bytes]" in run.output
+    assert "after" in run.output
+
+
+def test_an_error_mid_run_ends_the_whole_process_group(app, monkeypatch, tmp_path):
+    from steward.ansible import executor
+    from steward.models.ansible import AnsibleRunStatus
+    pidfile = tmp_path / "child.pid"
+    _use_script(monkeypatch, SPAWN_CHILD + "print('boom', flush=True)\ntime.sleep(60)\n",
+                str(pidfile))
+    real_scrub = executor.scrub
+
+    def scrub(line, secrets):
+        if line == "boom":
+            raise RuntimeError("reader failed")
+        return real_scrub(line, secrets)
+
+    monkeypatch.setattr(executor, "scrub", scrub)
+
+    async def _go():
+        run_id = await _new_run(app)
+        await asyncio.wait_for(
+            executor.start_run(app, run_id, "t.yml", "inv.ini", str(tmp_path)), 30)
+        return await _row(app, run_id)
+
+    run = asyncio.run(_go())
+    assert run.status == AnsibleRunStatus.failed
+    assert "reader failed" in run.output
+    assert _gone(int(pidfile.read_text()))
+
+
+def test_a_run_whose_task_is_cancelled_is_not_left_running(app, monkeypatch, tmp_path):
+    from steward.ansible import executor
+    from steward.models.ansible import AnsibleRunStatus
+    pidfile = tmp_path / "child.pid"
+    _use_script(monkeypatch, SPAWN_CHILD + "print('started', flush=True)\ntime.sleep(60)\n",
+                str(pidfile))
+
+    async def _go():
+        run_id = await _new_run(app)
+        task = asyncio.create_task(
+            executor.start_run(app, run_id, "t.yml", "inv.ini", str(tmp_path)))
+        deadline = time.monotonic() + 15
+        while "started" not in executor._run_lines.get(run_id, []):
+            assert time.monotonic() < deadline, "the run never started"
+            await asyncio.sleep(0.1)
+        task.cancel()                       # Steward shutting down mid-run
+        with pytest.raises(asyncio.CancelledError):
+            await task
+        return await _row(app, run_id)
+
+    run = asyncio.run(_go())
+    assert run.status == AnsibleRunStatus.cancelled
+    assert run.finished_at is not None
+    assert _gone(int(pidfile.read_text()))
+
+
+def test_a_cancel_landing_as_the_run_finishes_keeps_its_result(app, monkeypatch, tmp_path):
+    from steward.ansible import executor
+    from steward.models.ansible import AnsibleRunStatus
+    _use_script(monkeypatch, f"print('{RECAP}')")
+    real_scrub = executor.scrub
+    run_ids: list[str] = []
+
+    def scrub(line, secrets):
+        if line.startswith("localhost :"):   # the last line: the cancel arrives now
+            executor._cancelled.add(run_ids[0])
+        return real_scrub(line, secrets)
+
+    monkeypatch.setattr(executor, "scrub", scrub)
+
+    async def _go():
+        run_ids.append(await _new_run(app))
+        await executor.start_run(app, run_ids[0], "t.yml", "inv.ini", str(tmp_path))
+        return await _row(app, run_ids[0])
+
+    run = asyncio.run(_go())
+    assert run.status == AnsibleRunStatus.success