tests/integration/test_broker_concurrency_live.py
Rama main · réplica de lectura del repositorio original.
"""Real-execution validation for the 2026-07-01 audit concurrency fixes.
Unlike the fast API tests (in-process TestClient + in-memory SQLite, single event
loop, races *injected* via monkeypatch), this harness spins up a REAL broker
process and a REAL worker process talking over HTTP, and runs REAL subprocess
jobs. It's the "real-execution check at every system boundary" for the fixes
that only manifest across processes:
- H1 — a SIGTERM-ignoring workload is actually force-killed by the worker's
SIGKILL-escalation timer (never proven by TestClient, which runs no
subprocess and delivers no signals).
- cancel — a /cancel on a genuinely-running job actually terminates the child
and lands the job terminal (the cancel-latency + H3 signal path).
Gated behind JOBD_LIVE=1 so normal/CI runs skip it (it launches processes and
takes ~10-30s):
JOBD_LIVE=1 pytest tests/integration/test_broker_concurrency_live.py -v
The manual multi-worker partition scenario (M2 stale-worker rejection) is
described in tests/integration/README-live.md — it needs a real network
partition (or iptables drop) between a worker and the broker, which isn't
reliably automatable in-process.
"""
from __future__ import annotations
import json
import os
import signal
import socket
import subprocess
import sys
import tempfile
import time
import urllib.error
import urllib.request
from collections.abc import Iterator
from pathlib import Path
import pytest
# live marker + env skipif: see tests/test_e2e.py (audit 2026-07-05, L2).
pytestmark = [
pytest.mark.live,
pytest.mark.skipif(
os.environ.get("JOBD_LIVE") != "1",
reason="Set JOBD_LIVE=1 to run the live broker+worker integration harness",
),
]
# Run the code under test from THIS worktree, not whatever `jobd` is installed.
_SRC = str(Path(__file__).resolve().parents[2] / "src")
_PY = sys.executable
def _free_port() -> int:
s = socket.socket()
s.bind(("127.0.0.1", 0))
port = s.getsockname()[1]
s.close()
return port
class _Broker:
"""A real jobd broker subprocess over a temp DB + temp config."""
def __init__(self, tmp: Path):
self.tmp = tmp
self.port = _free_port()
self.base = f"http://127.0.0.1:{self.port}"
cfg = tmp / "config"
cfg.mkdir(parents=True, exist_ok=True)
(cfg / "projects.yaml").write_text(
"projects:\n project-a: { priority: 55 }\n _default: { priority: 40 }\n"
)
(cfg / "profiles.yaml").write_text(
"profiles:\n small: { vram_gb: 0, ram_gb: 1, cpus: 1, expected_runtime: 5m, "
"preemptible: true, host_hint: any, fast_path: true }\n"
)
(cfg / "classifier.yaml").write_text("rules: []\n")
self.env = dict(
os.environ,
PYTHONPATH=_SRC,
JOBD_ALLOW_NO_AUTH="1",
JOBD_DISABLE_TAILNET_ACL="1",
JOBD_CONFIG_DIR=str(cfg),
JOBD_DB_URL=f"sqlite:///{tmp}/jobd.db",
JOBD_HOST="127.0.0.1",
JOBD_PORT=str(self.port),
JOBD_LOGS_DIR=str(tmp / "logs"),
)
self.proc = subprocess.Popen(
[_PY, "-c", "from jobd.main import run; run()"],
env=self.env,
stdout=subprocess.PIPE,
stderr=subprocess.STDOUT,
)
self._await_health()
def _await_health(self) -> None:
for _ in range(100):
try:
if self.get("/health").get("status") == "ok":
return
except Exception:
time.sleep(0.1)
out = b""
if self.proc.stdout is not None:
self.proc.stdout.close()
raise RuntimeError(f"broker did not become healthy on {self.base}\n{out!r}")
def get(self, path: str) -> dict:
with urllib.request.urlopen(self.base + path, timeout=5) as r:
return json.load(r)
def post(self, path: str, body: dict, headers: dict | None = None) -> dict:
req = urllib.request.Request(
self.base + path,
data=json.dumps(body).encode(),
headers={"Content-Type": "application/json", **(headers or {})},
method="POST",
)
with urllib.request.urlopen(req, timeout=10) as r:
return json.load(r)
def post_status(self, path: str, body: dict, headers: dict | None = None) -> int:
"""POST and return the HTTP status code, without raising on 4xx/5xx —
for asserting the broker's stale-worker 409 rejection."""
req = urllib.request.Request(
self.base + path,
data=json.dumps(body).encode(),
headers={"Content-Type": "application/json", **(headers or {})},
method="POST",
)
try:
with urllib.request.urlopen(req, timeout=10) as r:
return r.status
except urllib.error.HTTPError as e:
return e.code
def stop(self) -> None:
_terminate(self.proc)
def _terminate(proc: subprocess.Popen) -> None:
if proc.poll() is not None:
return
proc.send_signal(signal.SIGTERM)
try:
proc.wait(timeout=15)
except subprocess.TimeoutExpired:
proc.kill()
proc.wait(timeout=5)
def _wait_state(broker: _Broker, jid: int, states: set[str], timeout_s: float) -> dict:
"""Poll until the job reaches one of `states`, or fail with the last snapshot."""
deadline = time.monotonic() + timeout_s
last: dict = {}
while time.monotonic() < deadline:
last = broker.get(f"/jobs/{jid}")
if last["state"] in states:
return last
time.sleep(0.25)
pytest.fail(f"job {jid} never reached {states} within {timeout_s}s; last={last}")
@pytest.fixture(scope="module")
def broker() -> Iterator[_Broker]:
tmp = Path(tempfile.mkdtemp(prefix="jobd-live-"))
b = _Broker(tmp)
try:
yield b
finally:
b.stop()
@pytest.fixture(scope="module")
def worker(broker: _Broker) -> Iterator[subprocess.Popen]:
env = dict(
broker.env,
JOBD_URL=broker.base,
JOBD_WORKER_HOST="live-w1",
# Short escalation grace so the SIGKILL-escalation test is fast.
JOBD_WORKER_WATCHDOG_KILL_GRACE_S="3",
)
proc = subprocess.Popen(
[_PY, "-c", "from jobd.worker.job_worker import main; main()"],
env=env,
stdout=subprocess.PIPE,
stderr=subprocess.STDOUT,
)
# Give the worker a moment to register its first heartbeat.
time.sleep(2.0)
try:
yield proc
finally:
_terminate(proc)
def _submit(broker: _Broker, cmd: list[str], **extra) -> int:
body = {"cmd": cmd, "cwd": "/tmp", "project": "project-a", **extra}
resp = broker.post("/submit", body)
assert "id" in resp, f"submit failed: {resp}"
return resp["id"]
def test_sanity_real_job_completes(broker: _Broker, worker: subprocess.Popen) -> None:
"""Baseline: a real job dispatched to a real worker runs and completes."""
jid = _submit(broker, ["bash", "-c", "echo hi; sleep 1; echo bye"])
final = _wait_state(broker, jid, {"completed", "failed"}, timeout_s=30)
assert final["state"] == "completed", final
assert final["exit_code"] == 0
def test_watchdog_escalates_to_sigkill_on_sigterm_ignoring_job(
broker: _Broker, worker: subprocess.Popen
) -> None:
"""H1 (real execution): a workload that installs `trap '' TERM` and then idles
must still be force-killed by the worker's SIGKILL-escalation timer once the
idle watchdog fires — not left RUNNING forever pinning its slot.
Before the fix the watchdog sent one SIGTERM and returned from poll_signals
with no kill timer; the stdout-read loop blocked forever on the ignore-TERM
child, so escalation never ran. This is exactly the case TestClient can't
exercise (no real subprocess, no real signals).
"""
# trap '' TERM => the process group ignores SIGTERM; only SIGKILL ends it.
# idle_timeout_s=2 => the idle watchdog fires ~2s after start (no output).
jid = _submit(
broker,
["bash", "-c", "trap '' TERM; while true; do sleep 30; done"],
idle_timeout_s=2,
)
# idle(2s) + poll interval + escalation grace(3s) + slack.
final = _wait_state(broker, jid, {"failed", "orphaned", "cancelled"}, timeout_s=40)
# The worker reports watchdog kills as failed with a termination_reason.
assert final["state"] in ("failed", "orphaned"), final
assert final.get("termination_reason") in ("idle_timeout", "wall_timeout"), final
def test_cancel_running_job_terminates_child(broker: _Broker, worker: subprocess.Popen) -> None:
"""A /cancel on a genuinely-running job must terminate the real child and
land the job terminal promptly (cancel-latency + the H3 signal path)."""
jid = _submit(broker, ["bash", "-c", "echo started; sleep 60"])
# Wait until it's actually running on the worker.
_wait_state(broker, jid, {"running", "assigned"}, timeout_s=15)
broker.post(f"/jobs/{jid}/cancel", {})
final = _wait_state(broker, jid, {"cancelled", "failed", "completed"}, timeout_s=20)
assert final["state"] == "cancelled", final
def test_stale_worker_reports_rejected_by_live_broker(
broker: _Broker, worker: subprocess.Popen
) -> None:
"""M2 (real HTTP): once a job is owned by one worker, the live broker must
refuse /complete and /log carrying a DIFFERENT `X-Jobd-Worker` (a stale
worker whose reclaimed job was re-dispatched) with 409, and must not let the
stale report terminal-ize or corrupt the running job. The owner's own report
is still accepted.
This validates the broker-side mechanism of M2 directly over real HTTP; the
full partition -> reclaim -> re-dispatch -> stale-completion sequence is the
manual procedure in README-live.md (it needs a real partition).
"""
jid = _submit(broker, ["bash", "-c", "echo started; sleep 60"])
running = _wait_state(broker, jid, {"running", "assigned"}, timeout_s=15)
owner = running["worker"]
assert owner, running
stale = {"X-Jobd-Worker": "ghost-worker-xyz"}
assert broker.post_status(f"/jobs/{jid}/complete", {"exit_code": 0}, stale) == 409
assert broker.post_status(f"/jobs/{jid}/log", {}, stale) == 409
# A stale terminal report must NOT have moved the job off running.
assert broker.get(f"/jobs/{jid}")["state"] in ("running", "assigned")
# The real owner can still terminate it (owner header matches job.worker).
assert broker.post_status(f"/jobs/{jid}/cancel", {}) == 200
_wait_state(broker, jid, {"cancelled", "failed", "completed"}, timeout_s=20)