tests/test_worker.py
Rama main · réplica de lectura del repositorio original.
"""Unit tests for the worker — pure functions + cancel-signal e2e with a real subprocess."""
import threading
import time
import jobd.worker.job_worker as job_worker
from jobd.worker.job_worker import (
_missing_launcher_path,
command_has_concurrent_ok,
compute_unregistered_vram,
effective_vram_request_gb_from_job,
gpu_admission_check,
pick_resource_snapshot_mock,
run_job,
)
def test_compute_unregistered_zero_when_all_tracked():
nvidia_procs = [(1000, 4096), (1001, 2048)] # (pid, MiB)
tracked_pids = {1000, 1001}
assert compute_unregistered_vram(nvidia_procs, tracked_pids) == 0.0
def test_compute_unregistered_counts_untracked():
nvidia_procs = [(1000, 4096), (9999, 8192)]
tracked_pids = {1000}
mib = 8192
expected_gb = round(mib / 1024.0, 2)
assert compute_unregistered_vram(nvidia_procs, tracked_pids) == expected_gb
def _stub_nvml(monkeypatch, free_gb: float, procs: list[tuple[int, int]]):
"""Force NVML on and stub the two callable seams."""
monkeypatch.setattr(job_worker, "_NVML_OK", True)
monkeypatch.setattr(job_worker, "nvidia_free_vram_gb", lambda: free_gb)
monkeypatch.setattr(job_worker, "nvidia_processes", lambda: procs)
def test_admission_check_no_gpu_required_returns_unblocked(monkeypatch):
"""required_gb<=0 short-circuits — non-GPU jobs don't gate on this."""
monkeypatch.setattr(job_worker, "_NVML_OK", True)
out = gpu_admission_check(0.0, tracked_pids=set())
assert out["blocked"] is False
assert out["required_gb"] == 0.0
def test_admission_check_nvml_unavailable_returns_unblocked(monkeypatch):
"""Hosts without NVML (e.g. server CPU-only) skip the gate cleanly."""
monkeypatch.setattr(job_worker, "_NVML_OK", False)
out = gpu_admission_check(8.0, tracked_pids=set())
assert out["blocked"] is False
assert out["foreign_pids"] == []
def test_admission_check_unblocked_when_free_meets_required(monkeypatch):
_stub_nvml(monkeypatch, free_gb=12.0, procs=[(1000, 4096)])
out = gpu_admission_check(8.0, tracked_pids={1000})
assert out["blocked"] is False
assert out["free_gb"] == 12.0
assert out["required_gb"] == 8.0
assert out["foreign_pids"] == [] # 1000 is owned by us
assert out["foreign_vram_gb"] == 0.0
def test_admission_check_blocked_when_foreign_holds_vram(monkeypatch):
"""Foreign holder pushes free below required — the load-bearing case."""
_stub_nvml(monkeypatch, free_gb=2.0, procs=[(1000, 4096), (9999, 10240)])
out = gpu_admission_check(8.0, tracked_pids={1000})
assert out["blocked"] is True
assert out["free_gb"] == 2.0
assert out["required_gb"] == 8.0
assert out["foreign_pids"] == [9999]
assert out["foreign_vram_gb"] == 10.0
def test_admission_check_foreign_diagnostics_even_when_unblocked(monkeypatch):
"""A foreign holder that doesn't push free below required: blocked=False
but the diagnostic fields still surface the foreign PID — useful for
co-routing telemetry later (#42).
"""
_stub_nvml(monkeypatch, free_gb=20.0, procs=[(9999, 8192)])
out = gpu_admission_check(8.0, tracked_pids=set())
assert out["blocked"] is False
assert out["foreign_pids"] == [9999]
assert out["foreign_vram_gb"] == 8.0
def test_admission_check_sums_multiple_foreign_holders(monkeypatch):
_stub_nvml(monkeypatch, free_gb=1.0, procs=[(8001, 5120), (8002, 6144)])
out = gpu_admission_check(8.0, tracked_pids=set())
assert out["blocked"] is True
assert out["foreign_pids"] == [8001, 8002] # sorted
assert out["foreign_vram_gb"] == 11.0
def test_effective_vram_explicit_wins():
"""#41d worker mirror: `vram_gb` overrides any tier-tag / gpu fallback."""
job = {"vram_gb": 16.0, "requires": {"gpu": True, "needs": ["cuda-8gb"]}}
assert effective_vram_request_gb_from_job(job) == 16.0
def test_effective_vram_tier_tag_picks_max():
job = {"requires": {"gpu": True, "needs": ["cuda-12gb", "cuda-32gb", "R"]}}
assert effective_vram_request_gb_from_job(job) == 32.0
def test_effective_vram_implicit_floor_when_only_gpu_true():
"""gpu=True but no explicit vram + no tier tag → 2 GB floor."""
job = {"requires": {"gpu": True, "needs": []}}
assert effective_vram_request_gb_from_job(job) == 2.0
def test_effective_vram_zero_for_non_gpu_job():
"""No requires.gpu, no tier tags, no explicit → 0 (gate is a no-op)."""
job = {"requires": {"gpu": False, "needs": []}}
assert effective_vram_request_gb_from_job(job) == 0.0
def test_effective_vram_handles_missing_requires():
"""Old job dicts without `requires` key shouldn't crash."""
assert effective_vram_request_gb_from_job({}) == 0.0
def test_effective_vram_malformed_tier_tag_ignored():
"""`cuda-bigboygb` falls through ValueError; valid tag still wins."""
job = {"requires": {"gpu": True, "needs": ["cuda-bigboygb", "cuda-12gb"]}}
assert effective_vram_request_gb_from_job(job) == 12.0
def test_concurrent_ok_marker_detected_in_bash_c():
"""`bash -c "python train.py # CONCURRENT_OK"` form."""
cmd = ["bash", "-c", "python train.py # CONCURRENT_OK"]
assert command_has_concurrent_ok(cmd) is True
def test_concurrent_ok_marker_detected_with_no_space():
cmd = ["bash", "-c", "python train.py #CONCURRENT_OK"]
assert command_has_concurrent_ok(cmd) is True
def test_concurrent_ok_marker_absent():
cmd = ["bash", "-c", "python train.py"]
assert command_has_concurrent_ok(cmd) is False
def test_concurrent_ok_marker_substring_does_not_false_match():
"""Word-boundary anchor: `CONCURRENT_OKAY` mustn't trip the bypass."""
cmd = ["bash", "-c", "echo CONCURRENT_OKAY"]
assert command_has_concurrent_ok(cmd) is False
def test_concurrent_ok_marker_handles_empty_cmd():
assert command_has_concurrent_ok(None) is False
assert command_has_concurrent_ok([]) is False
def test_pick_resource_snapshot_mock_shape():
snap = pick_resource_snapshot_mock()
for key in (
"host",
"free_vram_gb",
"unregistered_vram_gb",
"free_ram_gb",
"idle_cpus",
"host_aliases",
"arch",
"os",
"gpu",
"tags",
):
assert key in snap
class _FakeResp:
def __init__(self, status_code=200, data=None):
self.status_code = status_code
self._data = data or {}
def json(self):
return self._data
class _FakeClient:
"""Minimal stand-in for httpx.Client. Records POSTs, answers GETs.
cancel_after_s: wall-time elapsed from construction before /signal
endpoints return {"signal": "cancel"}.
"""
def __init__(self, cancel_after_s: float):
self.cancel_after_s = cancel_after_s
self.t0 = time.monotonic()
self.posts: list[tuple[str, dict]] = []
self.lock = threading.Lock()
def get(self, path, **kw):
if path.endswith("/signal"):
elapsed = time.monotonic() - self.t0
if elapsed >= self.cancel_after_s:
return _FakeResp(200, {"signal": "cancel"})
return _FakeResp(200, {"signal": None})
return _FakeResp(200, {})
def post(self, path, json=None, content=None, **kw):
with self.lock:
self.posts.append((path, json if json is not None else {"_raw": len(content or b"")}))
return _FakeResp(200, {})
def test_run_job_cancel_signal_terminates_child(tmp_path, monkeypatch):
"""run_job must honor /signal=cancel: SIGTERM the child, exit with
nonzero rc, and POST /complete with final_state=cancelled.
Real subprocess (bash -c sleep 10); fake client returns cancel after
1s. Whole test should finish in ~2s, well under the 10s sleep and
the 60s grace window.
Bypass the systemd-run wrap: the wrapper runs the workload in a
user-scope cgroup, and signal delivery via `systemctl --user kill`
is environment-dependent (CI containers often lack a user bus). Here
we care about the signal chain inside run_job, not the scope
integration — that's covered by tests/integration/test_cancel_latency.py.
"""
import jobd.worker.job_worker as job_worker
monkeypatch.setattr(job_worker.shutil, "which", lambda _name: None)
job = {"id": 9001, "cmd": ["bash", "-c", "sleep 10"], "cwd": str(tmp_path)}
client = _FakeClient(cancel_after_s=1.0)
tracked: set[int] = set()
t0 = time.monotonic()
run_job(client, job, tracked)
elapsed = time.monotonic() - t0
assert elapsed < 8.0, f"run_job should exit quickly after cancel, took {elapsed:.1f}s"
assert not tracked, "pid should be discarded after completion"
completes = [p for p in client.posts if p[0].endswith("/complete")]
assert len(completes) == 1, f"expected one /complete, got {client.posts}"
body = completes[0][1]
assert body["final_state"] == "cancelled"
assert body["exit_code"] != 0, f"cancelled process should have nonzero rc, got {body}"
def test_run_job_fast_path_skips_systemd_run_wrap(tmp_path, monkeypatch):
"""fast_path=true must bypass the systemd-run scope wrap even when
systemd-run exists on the worker. Regression guard: a fast-path job
wrapped in systemd-run --scope pays a ~300ms startup tax per invocation
and defeats the whole point."""
import subprocess
import jobd.worker.job_worker as job_worker
monkeypatch.setattr(
job_worker.shutil,
"which",
lambda name: f"/fake/bin/{name}" if name == "systemd-run" else None,
)
captured: dict[str, list[str]] = {}
orig_popen = subprocess.Popen
class _StubProc:
def __init__(self, cmd, **kw):
captured["cmd"] = cmd
self.pid = 12345
self.stdout = type("F", (), {"read": lambda _s, _n: b"", "close": lambda _s: None})()
def send_signal(self, _sig):
pass
def wait(self, timeout=None):
return 0
def kill(self):
pass
monkeypatch.setattr(subprocess, "Popen", _StubProc)
job = {"id": 9100, "cmd": ["echo", "hi"], "cwd": str(tmp_path), "fast_path": True}
run_job(_FakeClient(cancel_after_s=999.0), job, set())
assert captured["cmd"] == ["echo", "hi"], (
f"fast_path should skip systemd-run wrap, got {captured['cmd']}"
)
# sanity: suppress unused warning
_ = orig_popen
def test_run_job_non_fast_path_uses_named_scope(tmp_path, monkeypatch):
"""Non-fast-path jobs must be wrapped in `systemd-run --scope` with
a deterministic --unit name keyed off job id and explicit memory caps.
The scope name is what the cancel path signals via `systemctl --user
kill`, so it must match the worker's _signal_workload contract.
Regression for cancel-latency bug (2026-04-26): SIGTERM-to-pid was a
no-op because pid pointed at the systemd-run client, not at bash."""
import subprocess
import jobd.worker.job_worker as job_worker
monkeypatch.setattr(
job_worker.shutil,
"which",
lambda name: f"/fake/bin/{name}" if name == "systemd-run" else None,
)
captured: dict[str, list[str]] = {}
class _StubProc:
def __init__(self, cmd, **kw):
captured["cmd"] = cmd
self.pid = 12346
self.stdout = type("F", (), {"read": lambda _s, _n: b"", "close": lambda _s: None})()
def send_signal(self, _sig):
pass
def wait(self, timeout=None):
return 0
def kill(self):
pass
monkeypatch.setattr(subprocess, "Popen", _StubProc)
job = {"id": 9101, "cmd": ["python", "train.py"], "cwd": str(tmp_path), "fast_path": False}
run_job(_FakeClient(cancel_after_s=999.0), job, set())
assert captured["cmd"] == [
"/fake/bin/systemd-run",
"--user",
"--scope",
"--quiet",
"--same-dir",
f"--unit={job_worker.scope_unit_name(9101)}",
"-p",
"MemoryMax=14G",
"-p",
"MemorySwapMax=4G",
"--",
"python",
"train.py",
], f"unexpected wrap shape: {captured['cmd']}"
def test_run_job_no_systemd_run_runs_bare(tmp_path, monkeypatch):
"""When systemd-run is unavailable (CI containers, minimal images),
the worker must run the workload directly — no wrap, no crash."""
import subprocess
import jobd.worker.job_worker as job_worker
monkeypatch.setattr(job_worker.shutil, "which", lambda _name: None)
captured: dict[str, list[str]] = {}
class _StubProc:
def __init__(self, cmd, **kw):
captured["cmd"] = cmd
self.pid = 12347
self.stdout = type("F", (), {"read": lambda _s, _n: b"", "close": lambda _s: None})()
def send_signal(self, _sig):
pass
def wait(self, timeout=None):
return 0
def kill(self):
pass
monkeypatch.setattr(subprocess, "Popen", _StubProc)
job = {"id": 9102, "cmd": ["echo", "x"], "cwd": str(tmp_path), "fast_path": False}
run_job(_FakeClient(cancel_after_s=999.0), job, set())
assert captured["cmd"] == ["echo", "x"], f"expected bare cmd, got {captured['cmd']}"
def test_run_job_cancel_signals_scope_unit_not_pid(tmp_path, monkeypatch):
"""The cancel signal path must `systemctl --user kill --signal=TERM
jobd-{id}.scope` — NOT `proc.send_signal`. proc.pid points at the
systemd-run client, which exits as soon as the scope is set up;
signaling it is a silent no-op (root cause of the 2026-04-26 cancel-
latency bug). This test verifies the signaling channel."""
import subprocess
import jobd.worker.job_worker as job_worker
monkeypatch.setattr(
job_worker.shutil,
"which",
lambda name: f"/fake/bin/{name}" if name == "systemd-run" else None,
)
systemctl_calls: list[list[str]] = []
proc_signals: list[int] = []
cancel_seen = threading.Event()
class _StubStdout:
def __init__(self):
self._first = True
def read(self, _n):
# Block long enough for the signal-poll thread to observe cancel
# and call _signal_workload. Without this, the main loop ends
# before the poll thread gets scheduled.
if self._first:
self._first = False
cancel_seen.wait(timeout=3.0)
return b""
def close(self):
pass
class _StubProc:
def __init__(self, _cmd, **_kw):
self.pid = 22222
self.stdout = _StubStdout()
def send_signal(self, sig):
proc_signals.append(sig)
def wait(self, timeout=None):
return -15
def kill(self):
pass
orig_run = subprocess.run
def _fake_run(args, **kw):
if args and args[0] == "systemctl":
systemctl_calls.append(args)
cancel_seen.set()
class R:
returncode = 0
return R()
return orig_run(args, **kw)
monkeypatch.setattr(subprocess, "Popen", _StubProc)
monkeypatch.setattr(subprocess, "run", _fake_run)
job = {"id": 9201, "cmd": ["bash", "-c", "sleep 10"], "cwd": str(tmp_path)}
client = _FakeClient(cancel_after_s=0.0)
run_job(client, job, set())
assert any(
c[0] == "systemctl"
and "--user" in c
and "kill" in c
and c[-1] == job_worker.scope_unit_name(9201)
for c in systemctl_calls
), f"expected systemctl --user kill on scope, got {systemctl_calls}"
assert proc_signals == [], (
f"cancel must NOT fall through to proc.send_signal when scope is set, got {proc_signals}"
)
completes = [p for p in client.posts if p[0].endswith("/complete")]
assert completes and completes[0][1]["final_state"] == "cancelled"
def test_run_job_posts_started_after_popen(tmp_path, monkeypatch):
"""Right after subprocess.Popen returns, the worker must POST /started so
the broker can flip assigned -> running. Without this, jobs stay in
ASSIGNED for their entire run and MCP cancel signal_sent reads as null."""
import subprocess
import jobd.worker.job_worker as job_worker
monkeypatch.setattr(job_worker.shutil, "which", lambda _name: None)
class _StubProc:
def __init__(self, _cmd, **_kw):
self.pid = 12347
self.stdout = type("F", (), {"read": lambda _s, _n: b"", "close": lambda _s: None})()
def send_signal(self, _sig):
pass
def wait(self, timeout=None):
return 0
def kill(self):
pass
monkeypatch.setattr(subprocess, "Popen", _StubProc)
client = _FakeClient(cancel_after_s=999.0)
job = {"id": 9200, "cmd": ["echo", "hi"], "cwd": str(tmp_path)}
run_job(client, job, set())
started_posts = [p for p in client.posts if p[0] == "/jobs/9200/started"]
assert len(started_posts) == 1, (
f"expected exactly one /started POST, got {[p[0] for p in client.posts]}"
)
def _quiet_stub_proc_pair(rc_on_term: int = -15):
"""Build (StubProc, StubStdout) where stdout blocks until release_event
is set, then returns b"" (EOF). Lets the test choose when the workload
"finishes" — used to exercise watchdog-driven kills.
"""
import subprocess as _subprocess # noqa: F401
release_event = threading.Event()
class _StubStdout:
def read(self, _n):
release_event.wait(timeout=10.0)
return b""
def close(self):
pass
class _StubProc:
def __init__(self, _cmd, **_kw):
self.pid = 33333
self.stdout = _StubStdout()
def send_signal(self, _sig):
release_event.set()
def wait(self, timeout=None):
release_event.wait(timeout=10.0)
return rc_on_term
def kill(self):
release_event.set()
return _StubProc, release_event
def test_run_job_idle_timeout_kills_silent_workload(tmp_path, monkeypatch):
"""A job that produces zero output for idle_timeout_s seconds must be
killed via the scope cancel path and reported as failed with
termination_reason='idle_timeout'. This is the 2026-04-25 hung-project-c
failure mode: liveness != progress."""
import subprocess
import jobd.worker.job_worker as job_worker
monkeypatch.setattr(job_worker, "SIGNAL_POLL_INTERVAL_S", 0.05)
monkeypatch.setattr(
job_worker.shutil,
"which",
lambda name: f"/fake/bin/{name}" if name == "systemd-run" else None,
)
StubProc, release = _quiet_stub_proc_pair()
monkeypatch.setattr(subprocess, "Popen", StubProc)
systemctl_calls: list[list[str]] = []
def _fake_run(args, **_kw):
if args and args[0] == "systemctl":
systemctl_calls.append(args)
release.set()
class R:
returncode = 0
return R()
class R:
returncode = 0
return R()
monkeypatch.setattr(subprocess, "run", _fake_run)
job = {
"id": 9301,
"cmd": ["bash", "-c", "sleep 60"],
"cwd": str(tmp_path),
"idle_timeout_s": 1,
}
client = _FakeClient(cancel_after_s=999.0) # never user-cancel
t0 = time.monotonic()
run_job(client, job, set())
elapsed = time.monotonic() - t0
assert elapsed < 4.0, f"idle watchdog should fire promptly, took {elapsed:.1f}s"
assert any(
"kill" in c and c[-1] == job_worker.scope_unit_name(9301) for c in systemctl_calls
), f"expected systemctl kill on scope unit, got {systemctl_calls}"
completes = [p for p in client.posts if p[0].endswith("/complete")]
assert len(completes) == 1
body = completes[0][1]
assert body["final_state"] == "failed", (
f"watchdog timeout must surface as 'failed' so depends_on cascade fails children, got {body}"
)
assert body["termination_reason"] == "idle_timeout"
def test_run_job_max_wall_kills_long_running_workload(tmp_path, monkeypatch):
"""A job whose total runtime exceeds max_wall_s must be killed and
reported as failed with termination_reason='wall_timeout' — even if
it's producing output (i.e. not idle)."""
import subprocess
import jobd.worker.job_worker as job_worker
monkeypatch.setattr(job_worker, "SIGNAL_POLL_INTERVAL_S", 0.05)
monkeypatch.setattr(
job_worker.shutil,
"which",
lambda name: f"/fake/bin/{name}" if name == "systemd-run" else None,
)
StubProc, release = _quiet_stub_proc_pair()
monkeypatch.setattr(subprocess, "Popen", StubProc)
systemctl_calls: list[list[str]] = []
def _fake_run(args, **_kw):
if args and args[0] == "systemctl":
systemctl_calls.append(args)
release.set()
class R:
returncode = 0
return R()
class R:
returncode = 0
return R()
monkeypatch.setattr(subprocess, "run", _fake_run)
job = {
"id": 9302,
"cmd": ["bash", "-c", "yes"],
"cwd": str(tmp_path),
"max_wall_s": 1,
}
client = _FakeClient(cancel_after_s=999.0)
t0 = time.monotonic()
run_job(client, job, set())
elapsed = time.monotonic() - t0
assert elapsed < 4.0, f"wall watchdog should fire promptly, took {elapsed:.1f}s"
completes = [p for p in client.posts if p[0].endswith("/complete")]
assert len(completes) == 1
body = completes[0][1]
assert body["final_state"] == "failed"
assert body["termination_reason"] == "wall_timeout"
def test_run_job_env_default_idle_timeout(tmp_path, monkeypatch):
"""JOBD_WORKER_IDLE_TIMEOUT_S env var supplies a default when the job
doesn't specify idle_timeout_s. Lets ops set a fleet-wide watchdog
without changing every call site."""
import subprocess
import jobd.worker.job_worker as job_worker
monkeypatch.setenv("JOBD_WORKER_IDLE_TIMEOUT_S", "1")
monkeypatch.setattr(job_worker, "SIGNAL_POLL_INTERVAL_S", 0.05)
monkeypatch.setattr(job_worker.shutil, "which", lambda _name: None)
StubProc, release = _quiet_stub_proc_pair()
monkeypatch.setattr(subprocess, "Popen", StubProc)
def _fake_run(args, **_kw):
release.set()
class R:
returncode = 0
return R()
monkeypatch.setattr(subprocess, "run", _fake_run)
job = {"id": 9303, "cmd": ["bash", "-c", "sleep 60"], "cwd": str(tmp_path)}
client = _FakeClient(cancel_after_s=999.0)
t0 = time.monotonic()
run_job(client, job, set())
elapsed = time.monotonic() - t0
assert elapsed < 4.0, f"env-default idle watchdog should fire promptly, took {elapsed:.1f}s"
completes = [p for p in client.posts if p[0].endswith("/complete")]
assert len(completes) == 1
assert completes[0][1]["termination_reason"] == "idle_timeout"
def test_user_cancel_surfaces_as_cancelled_not_failed(tmp_path, monkeypatch):
"""Regression guard: the watchdog logic flips final_state from
'cancelled' to 'failed' only when termination_reason is set. A
user-initiated /cancel must still land as 'cancelled' so the manual-
abort surface stays clean."""
import subprocess
import jobd.worker.job_worker as job_worker
monkeypatch.setattr(job_worker, "SIGNAL_POLL_INTERVAL_S", 0.05)
monkeypatch.setattr(job_worker.shutil, "which", lambda _name: None)
StubProc, release = _quiet_stub_proc_pair()
monkeypatch.setattr(subprocess, "Popen", StubProc)
def _fake_run(_args, **_kw):
release.set()
class R:
returncode = 0
return R()
monkeypatch.setattr(subprocess, "run", _fake_run)
job = {"id": 9304, "cmd": ["bash", "-c", "sleep 5"], "cwd": str(tmp_path)}
client = _FakeClient(cancel_after_s=0.05) # cancel almost immediately
run_job(client, job, set())
completes = [p for p in client.posts if p[0].endswith("/complete")]
assert len(completes) == 1
body = completes[0][1]
assert body["final_state"] == "cancelled", body
# No termination_reason should be set for user cancels
assert "termination_reason" not in body or body.get("termination_reason") is None
def test_run_job_watchdog_escalates_to_sigkill_when_sigterm_ignored(tmp_path, monkeypatch):
"""A watchdog kill must escalate SIGTERM -> SIGKILL for a workload that
ignores SIGTERM and keeps stdout open. The wall/idle/first-output branches
route through _initiate_termination, which arms a kill timer; without it
(bare _signal_workload('TERM')) poll_signals returns, the stdout-read loop
blocks forever on the still-open pipe, and the post-loop
proc.wait(timeout=grace) escalation is never reached — the job pins its
slot (e.g. a GPU) indefinitely. Here only the SIGKILL lets the stub finish,
so reaching /complete promptly proves the timer fired."""
import subprocess
import jobd.worker.job_worker as job_worker
monkeypatch.setattr(job_worker, "SIGNAL_POLL_INTERVAL_S", 0.05)
monkeypatch.setattr(job_worker, "WATCHDOG_KILL_GRACE_S", 0.3)
monkeypatch.setattr(
job_worker.shutil,
"which",
lambda name: f"/fake/bin/{name}" if name == "systemd-run" else None,
)
# Released ONLY by the SIGKILL escalation. A SIGTERM leaves the workload
# alive with stdout still blocked, modelling the ignore-SIGTERM case.
killed = threading.Event()
class _StubStdout:
def read(self, _n):
killed.wait(timeout=10.0)
return b""
def close(self):
pass
class _StubProc:
def __init__(self, _cmd, **_kw):
self.pid = 34444
self.stdout = _StubStdout()
def send_signal(self, _sig):
pass
def poll(self):
return None if not killed.is_set() else -9
def wait(self, timeout=None):
killed.wait(timeout=10.0)
return -9
def kill(self):
killed.set()
monkeypatch.setattr(subprocess, "Popen", _StubProc)
systemctl_calls: list[list[str]] = []
def _fake_run(args, **_kw):
if args and args[0] == "systemctl":
systemctl_calls.append(args)
if any(a == "--signal=KILL" for a in args):
killed.set()
class R:
returncode = 0
return R()
monkeypatch.setattr(subprocess, "run", _fake_run)
job = {
"id": 9401,
"cmd": ["bash", "-c", "sleep 60"],
"cwd": str(tmp_path),
"idle_timeout_s": 1,
}
client = _FakeClient(cancel_after_s=999.0) # never user-cancel
t0 = time.monotonic()
run_job(client, job, set())
elapsed = time.monotonic() - t0
assert elapsed < 6.0, f"escalation should force-kill promptly, took {elapsed:.1f}s"
kill_calls = [
c
for c in systemctl_calls
if any(a == "--signal=KILL" for a in c) and c[-1] == job_worker.scope_unit_name(9401)
]
assert kill_calls, f"watchdog must escalate to SIGKILL on the scope unit, got {systemctl_calls}"
completes = [p for p in client.posts if p[0].endswith("/complete")]
assert len(completes) == 1
assert completes[0][1]["termination_reason"] == "idle_timeout"
def test_run_job_does_not_post_started_when_popen_fails(tmp_path, monkeypatch):
"""If Popen raises (bad cmd, missing executable), the worker reports
/complete with exit 127 — and must NOT POST /started, because there's
no live subprocess to be 'running'."""
import subprocess
import jobd.worker.job_worker as job_worker
monkeypatch.setattr(job_worker.shutil, "which", lambda _name: None)
def _boom(*_a, **_kw):
raise FileNotFoundError("nope")
monkeypatch.setattr(subprocess, "Popen", _boom)
client = _FakeClient(cancel_after_s=999.0)
job = {"id": 9201, "cmd": ["does-not-exist"], "cwd": str(tmp_path)}
run_job(client, job, set())
started_posts = [p for p in client.posts if p[0].endswith("/started")]
assert started_posts == [], f"unexpected /started POST after Popen failure: {client.posts}"
complete_posts = [p for p in client.posts if p[0].endswith("/complete")]
assert len(complete_posts) == 1
assert complete_posts[0][1]["exit_code"] == 127
class _PreemptClient:
"""FakeClient variant: returns signal=preempt after preempt_after_s.
Records all POSTs (raw bytes for /log, parsed JSON for everything
else) so tests can verify both the side effects of preempt handling
and the bodies of /complete and /checkpoint-complete.
"""
def __init__(self, preempt_after_s: float):
self.preempt_after_s = preempt_after_s
self.t0 = time.monotonic()
self.posts: list[tuple[str, object]] = []
self.lock = threading.Lock()
def get(self, path, **kw):
if path.endswith("/signal"):
elapsed = time.monotonic() - self.t0
if elapsed >= self.preempt_after_s:
return _FakeResp(200, {"signal": "preempt"})
return _FakeResp(200, {"signal": None})
return _FakeResp(200, {})
def post(self, path, json=None, content=None, **kw):
with self.lock:
payload: object = json if json is not None else (content or b"")
self.posts.append((path, payload))
return _FakeResp(200, {})
def test_run_job_preempt_with_checkpoint_token_emits_checkpoint_complete(tmp_path, monkeypatch):
"""#37: when the workload prints `jobd-checkpoint-complete` after
SIGTERM during a preempt, the worker POSTs /checkpoint-complete and
reports final_state=preempted."""
import jobd.worker.job_worker as job_worker
monkeypatch.setattr(job_worker.shutil, "which", lambda _name: None)
job = {
"id": 9301,
"cmd": [
"python3",
"-c",
"import signal,sys,time\n"
"signal.signal(signal.SIGTERM, lambda *a: ("
"print('jobd-checkpoint-complete', flush=True), sys.exit(0)))\n"
"time.sleep(30)\n",
],
"cwd": str(tmp_path),
"checkpoint_grace_s": 30,
}
client = _PreemptClient(preempt_after_s=1.0)
run_job(client, job, set())
paths = [p for p, _ in client.posts]
assert any(p.endswith("/checkpoint-complete") for p in paths), (
f"expected /checkpoint-complete POST, got {paths}"
)
completes = [
p
for p in client.posts
if p[0].endswith("/complete") and not p[0].endswith("/checkpoint-complete")
]
assert len(completes) == 1
assert completes[0][1]["final_state"] == "preempted"
def test_run_job_preempt_without_token_skips_checkpoint_complete(tmp_path, monkeypatch):
"""#37: a preempted workload that exits without printing the token
still completes as preempted, but no /checkpoint-complete POST fires."""
import jobd.worker.job_worker as job_worker
monkeypatch.setattr(job_worker.shutil, "which", lambda _name: None)
job = {
"id": 9302,
"cmd": ["bash", "-c", "sleep 30"],
"cwd": str(tmp_path),
}
client = _PreemptClient(preempt_after_s=1.0)
run_job(client, job, set())
paths = [p for p, _ in client.posts]
assert not any(p.endswith("/checkpoint-complete") for p in paths), (
f"unexpected /checkpoint-complete POST: {paths}"
)
completes = [p for p in client.posts if p[0].endswith("/complete")]
assert len(completes) == 1
assert completes[0][1]["final_state"] == "preempted"
def test_run_job_preempt_honors_checkpoint_grace_s(tmp_path, monkeypatch):
"""#37: an explicit checkpoint_grace_s overrides the worker's 60s
default — a workload that ignores SIGTERM gets SIGKILLed after the
configured grace window, not after a minute."""
import jobd.worker.job_worker as job_worker
monkeypatch.setattr(job_worker.shutil, "which", lambda _name: None)
job = {
"id": 9303,
"cmd": [
"python3",
"-c",
"import signal,time\nsignal.signal(signal.SIGTERM, lambda *a: None)\ntime.sleep(30)\n",
],
"cwd": str(tmp_path),
"checkpoint_grace_s": 2,
}
client = _PreemptClient(preempt_after_s=1.0)
t0 = time.monotonic()
run_job(client, job, set())
elapsed = time.monotonic() - t0
assert elapsed < 6.0, f"expected ~3s with grace=2, got {elapsed:.1f}s"
completes = [p for p in client.posts if p[0].endswith("/complete")]
assert completes[0][1]["final_state"] == "preempted"
def test_run_job_exports_checkpoint_grace_env_for_workload(tmp_path, monkeypatch):
"""#38: worker exposes the resolved per-job checkpoint_grace_s to the
workload via JOBD_CHECKPOINT_GRACE_S so install_preemption_handler
can compute time_remaining()."""
import jobd.worker.job_worker as job_worker
monkeypatch.setattr(job_worker.shutil, "which", lambda _name: None)
out_path = tmp_path / "env.txt"
job = {
"id": 9501,
"cmd": [
"bash",
"-c",
f"echo $JOBD_CHECKPOINT_GRACE_S > {out_path}",
],
"cwd": str(tmp_path),
"checkpoint_grace_s": 90,
}
run_job(_FakeClient(cancel_after_s=999.0), job, set())
assert out_path.read_text().strip() == "90"
def test_run_job_exports_default_checkpoint_grace_env_when_unset(tmp_path, monkeypatch):
"""No checkpoint_grace_s on the job → worker exports the default 60s
so install_preemption_handler always has a sane value to read."""
import jobd.worker.job_worker as job_worker
monkeypatch.setattr(job_worker.shutil, "which", lambda _name: None)
out_path = tmp_path / "env.txt"
job = {
"id": 9502,
"cmd": [
"bash",
"-c",
f"echo $JOBD_CHECKPOINT_GRACE_S > {out_path}",
],
"cwd": str(tmp_path),
}
run_job(_FakeClient(cancel_after_s=999.0), job, set())
assert out_path.read_text().strip() == "60"
# ---- #42 multi-tenant per-host co-routing -------------------------------
def _reset_in_flight():
with job_worker._in_flight_lock:
job_worker._in_flight.clear()
def test_max_concurrent_jobs_default_is_1(monkeypatch):
"""Default preserves single-slot behavior — opt-in via env."""
monkeypatch.delenv("JOBD_WORKER_MAX_CONCURRENT_JOBS", raising=False)
assert job_worker._max_concurrent_jobs() == 1
def test_max_concurrent_jobs_reads_env(monkeypatch):
monkeypatch.setenv("JOBD_WORKER_MAX_CONCURRENT_JOBS", "4")
assert job_worker._max_concurrent_jobs() == 4
def test_max_concurrent_jobs_floor_is_1(monkeypatch):
"""Garbage / negative env values fall back to 1, never 0."""
monkeypatch.setenv("JOBD_WORKER_MAX_CONCURRENT_JOBS", "0")
assert job_worker._max_concurrent_jobs() == 1
monkeypatch.setenv("JOBD_WORKER_MAX_CONCURRENT_JOBS", "-3")
assert job_worker._max_concurrent_jobs() == 1
monkeypatch.setenv("JOBD_WORKER_MAX_CONCURRENT_JOBS", "not-a-number")
assert job_worker._max_concurrent_jobs() == 1
def test_register_in_flight_records_ad():
_reset_in_flight()
try:
job_worker._register_in_flight({"id": 901, "vram_gb": 8.0, "ram_gb": 4.0, "cpus": 2})
v, r, c = job_worker._allocations_total()
assert v == 8.0
assert r == 4.0
assert c == 2
finally:
_reset_in_flight()
def test_register_in_flight_uses_effective_vram_for_tier_tag():
"""vram_gb=0 + cuda-32gb in needs → tier ladder fallback in the ad."""
_reset_in_flight()
try:
job_worker._register_in_flight(
{
"id": 902,
"vram_gb": 0,
"requires": {"needs": ["cuda-32gb"]},
"ram_gb": 0,
"cpus": 0,
}
)
v, _, _ = job_worker._allocations_total()
assert v == 32.0
finally:
_reset_in_flight()
def test_unregister_in_flight_clears_ad():
_reset_in_flight()
try:
job_worker._register_in_flight({"id": 903, "vram_gb": 8.0, "ram_gb": 4.0, "cpus": 2})
job_worker._unregister_in_flight(903)
assert job_worker._allocations_total() == (0.0, 0.0, 0)
finally:
_reset_in_flight()
def test_allocations_sum_across_multiple_in_flight():
_reset_in_flight()
try:
job_worker._register_in_flight({"id": 910, "vram_gb": 8.0, "ram_gb": 4.0, "cpus": 2})
job_worker._register_in_flight({"id": 911, "vram_gb": 12.0, "ram_gb": 6.0, "cpus": 3})
v, r, c = job_worker._allocations_total()
assert v == 20.0
assert r == 10.0
assert c == 5
finally:
_reset_in_flight()
def test_resource_snapshot_subtracts_in_flight_vram(monkeypatch):
"""Heartbeat must report adjusted (raw - in-flight) free, so the broker
matcher only sees true-available capacity. Without this, a worker
running one 24 GB job still advertises its full GPU as free."""
_reset_in_flight()
try:
monkeypatch.setattr(job_worker, "nvidia_free_vram_gb", lambda: 32.0)
monkeypatch.setattr(job_worker, "nvidia_processes", lambda: [])
job_worker._register_in_flight({"id": 920, "vram_gb": 24.0, "ram_gb": 0, "cpus": 0})
snap = job_worker.resource_snapshot(set())
assert snap["free_vram_gb"] == 8.0
finally:
_reset_in_flight()
def test_resource_snapshot_floor_at_zero_when_overcommitted(monkeypatch):
"""In-flight ad larger than current raw free (e.g., the workload
crashed and freed VRAM but the ad still reflects the matcher's
reservation) — clamp at 0 rather than going negative."""
_reset_in_flight()
try:
monkeypatch.setattr(job_worker, "nvidia_free_vram_gb", lambda: 4.0)
monkeypatch.setattr(job_worker, "nvidia_processes", lambda: [])
job_worker._register_in_flight({"id": 930, "vram_gb": 24.0, "ram_gb": 0, "cpus": 0})
snap = job_worker.resource_snapshot(set())
assert snap["free_vram_gb"] == 0.0
finally:
_reset_in_flight()
def test_resource_snapshot_no_in_flight_unchanged(monkeypatch):
"""No allocations → snapshot reports raw NVML readings unchanged."""
_reset_in_flight()
monkeypatch.setattr(job_worker, "nvidia_free_vram_gb", lambda: 30.0)
monkeypatch.setattr(job_worker, "nvidia_processes", lambda: [])
snap = job_worker.resource_snapshot(set())
assert snap["free_vram_gb"] == 30.0
def test_resource_snapshot_does_not_double_count_allocated_vram(monkeypatch):
"""M6: once an in-flight job has actually allocated its VRAM, NVML free
already reflects it — the ad reservation must NOT subtract it again, or a
steady multi-slot worker idles capacity the GPU actually has. Here a 24 GB
job on a 32 GB card (NVML free = 8) must still report 8 GB free, not
8 - 24 clamped to 0."""
_reset_in_flight()
try:
monkeypatch.setattr(job_worker, "nvidia_free_vram_gb", lambda: 8.0)
# The job's OWN GPU pid holds the full 24 GB (24 * 1024 MiB).
monkeypatch.setattr(job_worker, "nvidia_processes", lambda: [(4242, 24 * 1024)])
job_worker._register_in_flight({"id": 921, "vram_gb": 24.0, "ram_gb": 0, "cpus": 0})
snap = job_worker.resource_snapshot({4242}) # tracked_pids owns 4242
assert snap["free_vram_gb"] == 8.0
finally:
_reset_in_flight()
def test_resource_snapshot_reserves_only_unallocated_vram(monkeypatch):
"""A job mid-startup that has allocated only part of its ad reserves the
remainder: raw_free - max(0, ad - allocated). Here NVML free = 20, the job
has allocated 12 of its 24 GB ad, so reserve 12 -> report 8."""
_reset_in_flight()
try:
monkeypatch.setattr(job_worker, "nvidia_free_vram_gb", lambda: 20.0)
monkeypatch.setattr(job_worker, "nvidia_processes", lambda: [(4243, 12 * 1024)])
job_worker._register_in_flight({"id": 922, "vram_gb": 24.0, "ram_gb": 0, "cpus": 0})
snap = job_worker.resource_snapshot({4243})
assert snap["free_vram_gb"] == 8.0
finally:
_reset_in_flight()
def test_compute_owned_vram_sums_owned_pids_only():
"""compute_owned_vram is the mirror of compute_unregistered_vram: it sums
VRAM held by pids IN the owned set (GB, 2dp)."""
procs = [(100, 8 * 1024), (200, 4 * 1024), (300, 2 * 1024)]
assert job_worker.compute_owned_vram(procs, {100, 300}) == 10.0
assert job_worker.compute_owned_vram(procs, set()) == 0.0
assert job_worker.compute_owned_vram([], {100}) == 0.0
def test_resource_snapshot_reports_slot_usage(monkeypatch):
"""Heartbeat carries max_concurrent (the env knob) + running (live
in-flight count) so the broker can surface slot usage in `job workers`."""
_reset_in_flight()
try:
monkeypatch.setattr(job_worker, "nvidia_free_vram_gb", lambda: 30.0)
monkeypatch.setattr(job_worker, "nvidia_processes", lambda: [])
monkeypatch.setenv("JOBD_WORKER_MAX_CONCURRENT_JOBS", "3")
job_worker._register_in_flight({"id": 1, "vram_gb": 0, "ram_gb": 0, "cpus": 1})
job_worker._register_in_flight({"id": 2, "vram_gb": 0, "ram_gb": 0, "cpus": 1})
snap = job_worker.resource_snapshot(set())
assert snap["max_concurrent"] == 3
assert snap["running"] == 2
finally:
_reset_in_flight()
def test_resource_snapshot_reports_in_flight_job_ids(monkeypatch):
"""SIGTERM-drain Phase 2: every heartbeat carries the worker's in-flight
job ids so the broker can reconcile claims a restarted worker no longer
knows about (docs/runbook.md, "Drain / restart a worker"). Sorted for determinism."""
_reset_in_flight()
try:
monkeypatch.setattr(job_worker, "nvidia_free_vram_gb", lambda: 30.0)
monkeypatch.setattr(job_worker, "nvidia_processes", lambda: [])
snap = job_worker.resource_snapshot(set())
assert snap["in_flight_job_ids"] == []
job_worker._register_in_flight({"id": 972, "vram_gb": 0, "ram_gb": 0, "cpus": 1})
job_worker._register_in_flight({"id": 971, "vram_gb": 0, "ram_gb": 0, "cpus": 1})
snap = job_worker.resource_snapshot(set())
assert snap["in_flight_job_ids"] == [971, 972]
finally:
_reset_in_flight()
def test_effective_owned_pids_unions_scope_cgroup_pids(tmp_path, monkeypatch):
"""A scope-wrapped job's real CUDA pids live in its cgroup, not in
tracked_pids (which holds the systemd-run client pid). The owned set must
union both — read from a real cgroup.procs file."""
_reset_in_flight()
try:
scope_dir = tmp_path / job_worker.scope_unit_name(9300)
scope_dir.mkdir()
(scope_dir / "cgroup.procs").write_text("4242\n4243\n")
monkeypatch.setattr(job_worker, "_REAPER_OK", True)
monkeypatch.setattr(
job_worker._cgroup_walk,
"resolve_user_scope_path",
lambda unit: scope_dir if unit == job_worker.scope_unit_name(9300) else None,
)
job_worker._register_in_flight({"id": 9300, "vram_gb": 8, "ram_gb": 0, "cpus": 0})
# 999 = the systemd-run client pid we Popen'd; 4242/4243 = real workload.
assert job_worker._effective_owned_pids({999}) == {999, 4242, 4243}
finally:
_reset_in_flight()
def test_effective_owned_pids_no_scope_is_just_tracked(monkeypatch):
"""Fast-path / unwrapped jobs have no scope cgroup (resolve → None); the
owned set is then exactly tracked_pids (the real child is already there)."""
_reset_in_flight()
try:
monkeypatch.setattr(job_worker, "_REAPER_OK", True)
monkeypatch.setattr(job_worker._cgroup_walk, "resolve_user_scope_path", lambda unit: None)
job_worker._register_in_flight({"id": 9301, "vram_gb": 0, "ram_gb": 0, "cpus": 0})
assert job_worker._effective_owned_pids({777}) == {777}
finally:
_reset_in_flight()
def test_own_scope_job_not_counted_as_foreign_vram(tmp_path, monkeypatch):
"""The P3.3 bug: NVML reports a forked-child pid (inside the scope cgroup)
holding the VRAM, but tracked_pids only has the job's top-level pid — so the
worker's OWN child was mis-counted as foreign/unregistered. With the
effective-owned-pids union (cgroup is the ownership boundary) it reads as 0
unregistered. RTX 5090 real-exec confirmed NVML reports the child pid."""
_reset_in_flight()
try:
scope_dir = tmp_path / job_worker.scope_unit_name(9302)
scope_dir.mkdir()
(scope_dir / "cgroup.procs").write_text("5000\n") # the real CUDA pid
monkeypatch.setattr(job_worker, "_REAPER_OK", True)
monkeypatch.setattr(
job_worker._cgroup_walk,
"resolve_user_scope_path",
lambda unit: scope_dir if unit == job_worker.scope_unit_name(9302) else None,
)
monkeypatch.setattr(job_worker, "nvidia_processes", lambda: [(5000, 8192)])
monkeypatch.setattr(job_worker, "nvidia_free_vram_gb", lambda: 24.0)
job_worker._register_in_flight({"id": 9302, "vram_gb": 8, "ram_gb": 0, "cpus": 0})
# tracked_pids holds only the top-level pid 999; the forked CUDA child
# (5000) is owned via the cgroup, so it is NOT foreign.
snap = job_worker.resource_snapshot({999})
assert snap["unregistered_vram_gb"] == 0.0
finally:
_reset_in_flight()
def test_the_reparented_orphan_sweep_runs_only_for_a_solo_job(monkeypatch):
"""P3.5: the reparented-orphan /proc sweep must only run when this is the
sole in-flight job. With a concurrent job registered, the global sweep
could kill the other job's reparented fast-path descendant. The gate is
now atomic with the scan (audit 2026-09-02 L-4; see
tests/unit/test_orphan_sweep_solo_gate_is_atomic.py), so this drives the
helper and observes whether the scan ran."""
scans: list[set[int]] = []
monkeypatch.setattr(
job_worker._subreaper,
"sweep_and_kill_reparented_orphans",
lambda known: scans.append(set(known)) or [],
)
_reset_in_flight()
try:
job_worker._sweep_reparented_orphans_if_solo(960, {1})
assert len(scans) == 1 # nothing in flight
job_worker._register_in_flight({"id": 960, "vram_gb": 0, "ram_gb": 0, "cpus": 0})
job_worker._sweep_reparented_orphans_if_solo(960, {1})
assert len(scans) == 2 # just me
job_worker._register_in_flight({"id": 961, "vram_gb": 0, "ram_gb": 0, "cpus": 0})
job_worker._sweep_reparented_orphans_if_solo(960, {1})
assert len(scans) == 2 # a concurrent job exists: no scan
job_worker._unregister_in_flight(961)
job_worker._sweep_reparented_orphans_if_solo(960, {1})
assert len(scans) == 3 # back to solo
finally:
_reset_in_flight()
def test_reserve_and_dispatch_registers_before_returning_threaded():
"""Over-subscribe race regression (P3.4): the in-flight reservation must be
taken SYNCHRONOUSLY — visible the instant _reserve_and_dispatch returns,
before the worker thread has even run. If registration were deferred into
the thread (the pre-fix behavior), the poll loop's capacity gate could
re-poll against a stale count and oversubscribe past max_concurrent."""
_reset_in_flight()
started = threading.Event()
release = threading.Event()
def fake_run(job):
started.set()
release.wait(5) # keep the job "in flight" so we can observe it
job_worker._unregister_in_flight(int(job["id"]))
threads: list = []
try:
job_worker._reserve_and_dispatch(
{"id": 940, "vram_gb": 8.0, "ram_gb": 4.0, "cpus": 2},
job_threads=threads,
run_in_thread=fake_run,
)
# Reservation visible immediately — the gate would see it next poll.
with job_worker._in_flight_lock:
assert 940 in job_worker._in_flight
assert len(threads) == 1
assert started.wait(5), "worker thread never started"
release.set()
threads[0].join(5)
# Thread's finally unregisters once the job finishes.
with job_worker._in_flight_lock:
assert 940 not in job_worker._in_flight
finally:
release.set()
_reset_in_flight()
def test_reserve_and_dispatch_single_slot_also_threads():
"""SIGTERM-drain prerequisite: even at max_concurrent == 1 the job must run
in a worker thread, never inline in the poll loop. Inline execution parks
the main thread inside proc.stdout.read() for the whole job, so a drain
can't start until the job ends naturally (docs/runbook.md, "Drain / restart a worker")."""
_reset_in_flight()
started = threading.Event()
release = threading.Event()
ran: list = []
def fake_run(job):
started.set()
release.wait(5) # if this ran inline, _reserve_and_dispatch would hang
ran.append(int(job["id"]))
job_worker._unregister_in_flight(int(job["id"]))
threads: list = []
try:
job_worker._reserve_and_dispatch(
{"id": 941, "vram_gb": 8.0, "ram_gb": 4.0, "cpus": 2},
job_threads=threads,
run_in_thread=fake_run,
)
# Returned while the job is still running — therefore not inline.
assert len(threads) == 1
with job_worker._in_flight_lock:
assert 941 in job_worker._in_flight # reservation already visible
assert started.wait(5), "worker thread never started"
release.set()
threads[0].join(5)
assert ran == [941]
with job_worker._in_flight_lock:
assert 941 not in job_worker._in_flight
finally:
release.set()
_reset_in_flight()
def test_reserve_and_dispatch_caps_concurrent_reservations():
"""Driving N claims through _reserve_and_dispatch with a gate check between
each (mirroring the poll loop) never exceeds max_concurrent live
reservations — the synchronous registration closes the race window."""
_reset_in_flight()
release = threading.Event()
max_concurrent = 3
threads: list = []
def fake_run(job):
release.wait(5)
job_worker._unregister_in_flight(int(job["id"]))
try:
admitted = 0
for jid in range(950, 960): # 10 candidate jobs, cap is 3
with job_worker._in_flight_lock:
live = len(job_worker._in_flight)
if live >= max_concurrent:
continue # gate refuses, exactly as the poll loop does
job_worker._reserve_and_dispatch(
{"id": jid, "vram_gb": 8.0, "ram_gb": 4.0, "cpus": 2},
job_threads=threads,
run_in_thread=fake_run,
)
admitted += 1
with job_worker._in_flight_lock:
assert len(job_worker._in_flight) <= max_concurrent
assert admitted == max_concurrent
release.set()
for t in threads:
t.join(5)
finally:
release.set()
_reset_in_flight()
def test_run_job_exposes_checkpoint_dir(tmp_path, monkeypatch):
"""run_job must set JOBD_CHECKPOINT_DIR in the spawned env, pointing at a
pre-created directory under the default root (~/.local/share/jobd/checkpoints).
Workloads use this to write durable checkpoints during a preempt."""
import subprocess
import jobd.worker.job_worker as job_worker
monkeypatch.setattr(
job_worker.shutil,
"which",
lambda name: f"/fake/bin/{name}" if name == "systemd-run" else None,
)
# Force the default root: clear override, redirect HOME to tmp so we don't
# touch the real ~/.local/share.
monkeypatch.delenv("JOBD_WORKER_CHECKPOINT_ROOT", raising=False)
monkeypatch.delenv("XDG_DATA_HOME", raising=False)
monkeypatch.setenv("HOME", str(tmp_path))
captured: dict[str, dict] = {}
class _StubProc:
def __init__(self, cmd, **kw):
captured["env"] = kw.get("env") or {}
self.pid = 12345
self.stdout = type("F", (), {"read": lambda _s, _n: b"", "close": lambda _s: None})()
def send_signal(self, _sig):
pass
def wait(self, timeout=None):
return 0
def kill(self):
pass
monkeypatch.setattr(subprocess, "Popen", _StubProc)
job = {
"id": 9200,
"cmd": ["echo", "hi"],
"cwd": str(tmp_path),
"fast_path": True,
}
run_job(_FakeClient(cancel_after_s=999.0), job, set())
ckpt_dir = captured["env"].get("JOBD_CHECKPOINT_DIR")
assert ckpt_dir, f"JOBD_CHECKPOINT_DIR not set; env keys: {sorted(captured['env'].keys())}"
expected_root = tmp_path / ".local" / "share" / "jobd" / "checkpoints"
assert ckpt_dir == str(expected_root / "9200"), (
f"expected {expected_root / '9200'}, got {ckpt_dir}"
)
assert (expected_root / "9200").is_dir(), f"directory not created at {ckpt_dir}"
def test_run_job_propagates_submitted_env(tmp_path, monkeypatch):
"""P0.3 (CRIT M1): caller-submitted env vars (`job["env"]`) reach the
workload subprocess. They layer over the worker's inherited environment,
and jobd-internal JOBD_* vars still take precedence (a caller cannot
clobber JOBD_CHECKPOINT_DIR)."""
import subprocess
import jobd.worker.job_worker as job_worker
monkeypatch.setattr(
job_worker.shutil,
"which",
lambda name: f"/fake/bin/{name}" if name == "systemd-run" else None,
)
monkeypatch.delenv("JOBD_WORKER_CHECKPOINT_ROOT", raising=False)
monkeypatch.delenv("XDG_DATA_HOME", raising=False)
monkeypatch.setenv("HOME", str(tmp_path))
captured: dict[str, dict] = {}
class _StubProc:
def __init__(self, cmd, **kw):
captured["env"] = kw.get("env") or {}
self.pid = 12346
self.stdout = type("F", (), {"read": lambda _s, _n: b"", "close": lambda _s: None})()
def send_signal(self, _sig):
pass
def wait(self, timeout=None):
return 0
def kill(self):
pass
monkeypatch.setattr(subprocess, "Popen", _StubProc)
job = {
"id": 9300,
"cmd": ["printenv"],
"cwd": str(tmp_path),
"fast_path": True,
"env": {
"MY_TOKEN": "abc123",
# A caller MUST NOT be able to clobber jobd-internal vars.
"JOBD_CHECKPOINT_DIR": "/tmp/evil",
},
}
run_job(_FakeClient(cancel_after_s=999.0), job, set())
assert captured["env"].get("MY_TOKEN") == "abc123"
# jobd-internal var wins over a caller attempting to override it.
assert captured["env"].get("JOBD_CHECKPOINT_DIR") != "/tmp/evil"
assert captured["env"]["JOBD_CHECKPOINT_DIR"].endswith("/9300")
def test_run_job_scrubs_broker_token_from_workload_env(tmp_path, monkeypatch):
"""The worker's JOBD_API_TOKEN must not be inherited by workloads — any
script or framework inside the job can dump its environment, and the
shared broker credential would ride along. A job that genuinely needs it
opts in by passing it in its own submitted `env`, which layers over the
scrub. Ordinary worker-side env vars are still inherited."""
import subprocess
monkeypatch.setenv("JOBD_API_TOKEN", "worker-secret-token")
monkeypatch.setenv("SOME_WORKER_VAR", "inherited-ok")
monkeypatch.setenv("HOME", str(tmp_path))
captured: dict[str, dict] = {}
class _StubProc:
def __init__(self, cmd, **kw):
captured["env"] = kw.get("env") or {}
self.pid = 12347
self.stdout = type("F", (), {"read": lambda _s, _n: b"", "close": lambda _s: None})()
def send_signal(self, _sig):
pass
def wait(self, timeout=None):
return 0
def kill(self):
pass
monkeypatch.setattr(subprocess, "Popen", _StubProc)
job = {"id": 9400, "cmd": ["printenv"], "cwd": str(tmp_path), "fast_path": True}
run_job(_FakeClient(cancel_after_s=999.0), job, set())
assert "JOBD_API_TOKEN" not in captured["env"], (
"the broker credential leaked into the workload environment"
)
assert captured["env"].get("SOME_WORKER_VAR") == "inherited-ok"
# Explicit opt-in: a submitted env carrying the token IS delivered.
job = {
"id": 9401,
"cmd": ["printenv"],
"cwd": str(tmp_path),
"fast_path": True,
"env": {"JOBD_API_TOKEN": "job-scoped-token"},
}
run_job(_FakeClient(cancel_after_s=999.0), job, set())
assert captured["env"].get("JOBD_API_TOKEN") == "job-scoped-token"
def test_run_job_checkpoint_dir_root_override(tmp_path, monkeypatch):
"""JOBD_WORKER_CHECKPOINT_ROOT overrides the default root. Used when an
operator wants checkpoints on a different filesystem (e.g., a fast NVMe
instead of HDD-backed home)."""
import subprocess
import jobd.worker.job_worker as job_worker
monkeypatch.setattr(
job_worker.shutil,
"which",
lambda name: f"/fake/bin/{name}" if name == "systemd-run" else None,
)
override_root = tmp_path / "fast-disk" / "jobd-ckpts"
monkeypatch.setenv("JOBD_WORKER_CHECKPOINT_ROOT", str(override_root))
captured: dict[str, dict] = {}
class _StubProc:
def __init__(self, cmd, **kw):
captured["env"] = kw.get("env") or {}
self.pid = 12346
self.stdout = type("F", (), {"read": lambda _s, _n: b"", "close": lambda _s: None})()
def send_signal(self, _sig):
pass
def wait(self, timeout=None):
return 0
def kill(self):
pass
monkeypatch.setattr(subprocess, "Popen", _StubProc)
job = {
"id": 9201,
"cmd": ["echo", "hi"],
"cwd": str(tmp_path),
"fast_path": True,
}
run_job(_FakeClient(cancel_after_s=999.0), job, set())
ckpt_dir = captured["env"]["JOBD_CHECKPOINT_DIR"]
assert ckpt_dir == str(override_root / "9201"), (
f"override not honored: expected {override_root / '9201'}, got {ckpt_dir}"
)
assert (override_root / "9201").is_dir()
def test_run_job_checkpoint_dir_override_expands_tilde(tmp_path, monkeypatch):
"""JOBD_WORKER_CHECKPOINT_ROOT supports ~-prefixed paths via expanduser,
matching the default branch. Operators setting the env var via systemd
Environment= (where shell expansion doesn't happen) should not get a
literal ~/ckpts directory."""
import subprocess
import jobd.worker.job_worker as job_worker
monkeypatch.setattr(
job_worker.shutil,
"which",
lambda name: f"/fake/bin/{name}" if name == "systemd-run" else None,
)
fake_home = tmp_path / "home"
fake_home.mkdir()
monkeypatch.setenv("HOME", str(fake_home))
monkeypatch.setenv("JOBD_WORKER_CHECKPOINT_ROOT", "~/ckpts-override")
captured: dict[str, dict] = {}
class _StubProc:
def __init__(self, cmd, **kw):
captured["env"] = kw.get("env") or {}
self.pid = 12347
self.stdout = type("F", (), {"read": lambda _s, _n: b"", "close": lambda _s: None})()
def send_signal(self, _sig):
pass
def wait(self, timeout=None):
return 0
def kill(self):
pass
monkeypatch.setattr(subprocess, "Popen", _StubProc)
job = {
"id": 9202,
"cmd": ["echo", "hi"],
"cwd": str(tmp_path),
"fast_path": True,
}
run_job(_FakeClient(cancel_after_s=999.0), job, set())
ckpt_dir = captured["env"]["JOBD_CHECKPOINT_DIR"]
expected = fake_home / "ckpts-override" / "9202"
assert ckpt_dir == str(expected), (
f"~ not expanded in override: expected {expected}, got {ckpt_dir}"
)
# Sanity: literal-~ path should NOT exist
assert not (tmp_path / "~" / "ckpts-override").exists()
assert expected.is_dir()
def test_run_job_checkpoint_dir_xdg_data_home_branch(tmp_path, monkeypatch):
"""XDG_DATA_HOME (when set and non-empty) wins over the ~/.local/share
fallback. Closes the contract→test gap on resolution rule #2."""
import subprocess
import jobd.worker.job_worker as job_worker
monkeypatch.setattr(
job_worker.shutil,
"which",
lambda name: f"/fake/bin/{name}" if name == "systemd-run" else None,
)
monkeypatch.delenv("JOBD_WORKER_CHECKPOINT_ROOT", raising=False)
xdg_root = tmp_path / "xdg-data"
monkeypatch.setenv("XDG_DATA_HOME", str(xdg_root))
monkeypatch.setenv("HOME", str(tmp_path / "home-should-not-be-used"))
captured: dict[str, dict] = {}
class _StubProc:
def __init__(self, cmd, **kw):
captured["env"] = kw.get("env") or {}
self.pid = 12348
self.stdout = type("F", (), {"read": lambda _s, _n: b"", "close": lambda _s: None})()
def send_signal(self, _sig):
pass
def wait(self, timeout=None):
return 0
def kill(self):
pass
monkeypatch.setattr(subprocess, "Popen", _StubProc)
job = {
"id": 9203,
"cmd": ["echo", "hi"],
"cwd": str(tmp_path),
"fast_path": True,
}
run_job(_FakeClient(cancel_after_s=999.0), job, set())
ckpt_dir = captured["env"]["JOBD_CHECKPOINT_DIR"]
expected = xdg_root / "jobd" / "checkpoints" / "9203"
assert ckpt_dir == str(expected), (
f"XDG_DATA_HOME branch not taken: expected {expected}, got {ckpt_dir}"
)
assert expected.is_dir()
def test_missing_launcher_path_absolute_missing(tmp_path):
missing = tmp_path / "nonexistent.sh"
assert _missing_launcher_path([str(missing), "arg"], cwd=str(tmp_path)) == str(missing)
def test_missing_launcher_path_relative_missing(tmp_path):
out = _missing_launcher_path(["./nope.sh"], cwd=str(tmp_path))
assert out == str(tmp_path / "nope.sh")
def test_missing_launcher_path_existing_executable_is_none(tmp_path):
script = tmp_path / "ok.sh"
script.write_text("#!/bin/sh\necho hi\n")
script.chmod(0o755)
assert _missing_launcher_path([str(script)], cwd=str(tmp_path)) is None
def test_missing_launcher_path_existing_non_executable_caught(tmp_path):
script = tmp_path / "noexec.sh"
script.write_text("hi\n")
script.chmod(0o644)
assert _missing_launcher_path([str(script)], cwd=str(tmp_path)) == str(script)
def test_missing_launcher_path_path_resolved_name_skipped(tmp_path):
# `bash`, `python`, `ctest` etc. — Popen resolves these via PATH; we
# intentionally don't pre-check them. Returns None even though no such
# absolute file exists.
assert _missing_launcher_path(["bash", "./inner.sh"], cwd=str(tmp_path)) is None
assert _missing_launcher_path(["python", "-c", "print(1)"], cwd=str(tmp_path)) is None
def test_missing_launcher_path_empty_cmd():
assert _missing_launcher_path([], cwd="/tmp") is None
def test_run_job_refuses_dispatch_when_launcher_missing(tmp_path, monkeypatch):
"""Regression: jobs 261/262 (2026-04-30) failed exit 127 because the
launcher wasn't committed. Worker must surface this at dispatch with
termination_reason=launcher_missing rather than exec-ing into a silent
failure inside the systemd-run scope."""
import subprocess
monkeypatch.setattr(job_worker.shutil, "which", lambda _name: None)
popen_calls: list[list[str]] = []
def _no_popen(cmd, **kw):
popen_calls.append(cmd)
raise AssertionError(f"Popen must not be called when launcher is missing; got {cmd}")
monkeypatch.setattr(subprocess, "Popen", _no_popen)
missing_path = str(tmp_path / "never_committed.sh")
job = {"id": 9999, "cmd": [missing_path, "--arg"], "cwd": str(tmp_path)}
client = _FakeClient(cancel_after_s=999.0)
run_job(client, job, set())
assert popen_calls == [], "Popen should not be reached for a missing launcher"
completes = [p for p in client.posts if p[0].endswith("/complete")]
assert len(completes) == 1, f"expected one /complete, got posts={client.posts}"
body = completes[0][1]
assert body["final_state"] == "failed"
assert body["exit_code"] == 127
assert body["termination_reason"] == "launcher_missing"
# The offending path should also be logged via /log for diagnostics.
# (_FakeClient stores content-POSTs as {"_raw": byte_length}; we just
# check that a non-empty log line landed.)
log_posts = [p for p in client.posts if p[0].endswith("/log")]
assert len(log_posts) == 1
assert log_posts[0][1].get("_raw", 0) > 0
def test_run_job_first_output_timeout_kills_silent_start(tmp_path, monkeypatch):
"""A job that produces zero output within
JOBD_WORKER_FIRST_OUTPUT_TIMEOUT_S of dispatch must be killed via
the scope cancel path and reported as failed with
termination_reason='first_output_timeout'. Distinct from idle_timeout
in that it has a typically-shorter, fires-once-only deadline."""
import subprocess
monkeypatch.setenv("JOBD_WORKER_FIRST_OUTPUT_TIMEOUT_S", "1")
monkeypatch.setattr(job_worker, "SIGNAL_POLL_INTERVAL_S", 0.05)
monkeypatch.setattr(
job_worker.shutil,
"which",
lambda name: f"/fake/bin/{name}" if name == "systemd-run" else None,
)
StubProc, release = _quiet_stub_proc_pair()
monkeypatch.setattr(subprocess, "Popen", StubProc)
systemctl_calls: list[list[str]] = []
def _fake_run(args, **_kw):
if args and args[0] == "systemctl":
systemctl_calls.append(args)
release.set()
class R:
returncode = 0
return R()
class R:
returncode = 0
return R()
monkeypatch.setattr(subprocess, "run", _fake_run)
job = {"id": 9401, "cmd": ["bash", "-c", "sleep 60"], "cwd": str(tmp_path)}
client = _FakeClient(cancel_after_s=999.0)
t0 = time.monotonic()
run_job(client, job, set())
elapsed = time.monotonic() - t0
assert elapsed < 4.0, f"first-output watchdog should fire promptly, took {elapsed:.1f}s"
assert any(
"kill" in c and c[-1] == job_worker.scope_unit_name(9401) for c in systemctl_calls
), f"expected systemctl kill on scope unit, got {systemctl_calls}"
completes = [p for p in client.posts if p[0].endswith("/complete")]
assert len(completes) == 1
body = completes[0][1]
assert body["final_state"] == "failed"
assert body["termination_reason"] == "first_output_timeout"
def test_run_job_first_output_timeout_disarms_after_first_byte(tmp_path, monkeypatch):
"""Once the workload produces a single byte, the first-output watchdog
must disarm permanently — subsequent silence is the idle_timeout's job.
Without this disarming, the watchdog would double-fire and kill jobs
that legitimately quiet down after a startup banner."""
import subprocess
monkeypatch.setenv("JOBD_WORKER_FIRST_OUTPUT_TIMEOUT_S", "1")
monkeypatch.setattr(job_worker, "SIGNAL_POLL_INTERVAL_S", 0.05)
monkeypatch.setattr(job_worker.shutil, "which", lambda _name: None) # fast-path
# Stub stdout: one chunk immediately, then EOF after a short wait.
yielded = threading.Event()
finished = threading.Event()
class _StubStdout:
def __init__(self):
self._yielded = False
def read(self, _n):
if not self._yielded:
self._yielded = True
yielded.set()
return b"startup banner\n"
finished.wait(timeout=3.0)
return b""
def close(self):
pass
class _StubProc:
def __init__(self, _cmd, **_kw):
self.pid = 44444
self.stdout = _StubStdout()
def send_signal(self, _sig):
finished.set()
def wait(self, timeout=None):
finished.wait(timeout=3.0)
return 0
def kill(self):
finished.set()
monkeypatch.setattr(subprocess, "Popen", _StubProc)
job = {"id": 9402, "cmd": ["echo", "x"], "cwd": str(tmp_path), "fast_path": True}
client = _FakeClient(cancel_after_s=999.0)
# Let the first-output watchdog window elapse — but the byte from
# _StubStdout lands first, so disarm should fire and the workload runs
# to clean EOF.
def _release_after_disarm():
yielded.wait(timeout=3.0)
time.sleep(2.0) # well past first_output_timeout_s=1
finished.set()
t = threading.Thread(target=_release_after_disarm, daemon=True)
t.start()
run_job(client, job, set())
completes = [p for p in client.posts if p[0].endswith("/complete")]
assert len(completes) == 1
body = completes[0][1]
# No watchdog fired → either no termination_reason, or definitely not
# first_output_timeout.
assert body.get("termination_reason") != "first_output_timeout"
assert body["final_state"] in ("completed", "failed") # depends on rc
def test_run_job_refuses_admission_when_cwd_missing():
"""B: a cwd absent on this host -> POST refuse-admission(cwd_missing), no
subprocess, no exit-127 /complete."""
posts = []
class _RecordingClient:
def post(self, path, json=None, content=None, **kw):
posts.append((path, json))
class _R:
status_code = 200
def json(self_inner):
return {}
return _R()
job = {"id": 7, "cmd": ["true"], "cwd": "/no/such/dir/xyz_nonexistent", "env": {}}
run_job(_RecordingClient(), job, set())
paths = [p for p, _ in posts]
assert any(p == "/jobs/7/refuse-admission" for p in paths), paths
body = next(j for p, j in posts if p == "/jobs/7/refuse-admission")
assert body["reason"] == "cwd_missing"
assert body["cwd"] == "/no/such/dir/xyz_nonexistent"
# must NOT have started the workload / posted a 127 completion
assert not any(p == "/jobs/7/complete" for p in paths), paths