test(issue-14): decouple the disk-heartbeat wrapper test from broker latency
The test blocked on each publish_event heartbeat, and against the public
broker a publish (with retries) takes 2-6 s, so consecutive events.log
records landed further apart than the 6 s idle window and the wrapper
closed the job before the next heartbeat - a broker-latency flake that
failed the full suite once and passed on re-run.
Heartbeats are now fired without waiting for the wire publish (the disk
record is written before the broker is touched, ebe58f6) and keep coming
for 20 s to outlast a slow connect; the wall clock is widened to match.
3/3 green in a row afterwards.
Refs #14
Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01VCgu9Fa9iT7TbJUSzdq4Ff
This commit is contained in:
1 parent
9d52bb7024
commit
fbe2731980
1 file changed
+22
-9
@@ -326,7 +326,7 @@ def test_wrapper_idle_timeout_grace_accepts_disk_heartbeat(tmp_path):
|
|||||||
"--prompt", "issue 14 disk heartbeat",
|
"--prompt", "issue 14 disk heartbeat",
|
||||||
"--registry-dir", str(reg_dir),
|
"--registry-dir", str(reg_dir),
|
||||||
"--workdir", str(tmp_path),
|
"--workdir", str(tmp_path),
|
||||||
"--timeout", "60",
|
"--timeout", "90",
|
||||||
"--idle-timeout", "6",
|
"--idle-timeout", "6",
|
||||||
],
|
],
|
||||||
cwd=str(tmp_path), env=env, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True,
|
cwd=str(tmp_path), env=env, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True,
|
||||||
@@ -341,20 +341,33 @@ def test_wrapper_idle_timeout_grace_accepts_disk_heartbeat(tmp_path):
|
|||||||
time.sleep(0.1)
|
time.sleep(0.1)
|
||||||
assert jid, "job was not registered"
|
assert jid, "job was not registered"
|
||||||
|
|
||||||
|
# Heartbeats are fired without waiting for the wire publish: publish_event
|
||||||
|
# records the event in events.log before it touches the broker (ebe58f6),
|
||||||
|
# and that on-disk timestamp is what the wrapper's grace reads. Blocking on
|
||||||
|
# each publish (2-6 s against the public broker, with retries) spaced the
|
||||||
|
# disk records further apart than the 6 s idle window and the job was
|
||||||
|
# closed before the next heartbeat landed - a broker-latency flake, not a
|
||||||
|
# wrapper defect. An agent's own heartbeat timer behaves the same way.
|
||||||
|
beats = []
|
||||||
|
|
||||||
def heartbeat(detail: str):
|
def heartbeat(detail: str):
|
||||||
subprocess.run(
|
beats.append(subprocess.Popen(
|
||||||
[sys.executable, str(PUBLISH_PY), "--registry-dir", str(reg_dir),
|
[sys.executable, str(PUBLISH_PY), "--registry-dir", str(reg_dir),
|
||||||
"--job", jid, "--event", "progress", "--detail", detail],
|
"--job", jid, "--event", "progress", "--detail", detail],
|
||||||
capture_output=True, text=True, cwd=str(tmp_path), env=env,
|
stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, cwd=str(tmp_path), env=env,
|
||||||
)
|
))
|
||||||
|
|
||||||
# Subscriber idles out at ~6 s (plus broker connect); heartbeats then keep
|
# Subscriber idles out at ~6 s plus the broker connect (the public broker
|
||||||
# arriving 2 s apart, well inside the 6 s window each one re-opens (the
|
# has taken 5-10 s); heartbeats then keep landing 2 s apart for 20 s, well
|
||||||
# wrapper's poll iteration costs ~2 s of herdr RPCs, so the ratio matters).
|
# inside the 6 s window each one re-opens (the wrapper's poll iteration
|
||||||
|
# costs ~2 s of herdr RPCs, so the ratio matters) and long enough to
|
||||||
|
# outlast a slow connect.
|
||||||
time.sleep(9)
|
time.sleep(9)
|
||||||
for i in range(5):
|
for i in range(10):
|
||||||
heartbeat(f"still working {i}")
|
heartbeat(f"still working {i}")
|
||||||
time.sleep(2)
|
time.sleep(2)
|
||||||
|
for b in beats:
|
||||||
|
b.wait(timeout=60)
|
||||||
done = subprocess.run(
|
done = subprocess.run(
|
||||||
[sys.executable, str(REGISTRY_PY), "--registry-dir", str(reg_dir),
|
[sys.executable, str(REGISTRY_PY), "--registry-dir", str(reg_dir),
|
||||||
"update", "--job", jid, "--status", "completed"],
|
"update", "--job", jid, "--status", "completed"],
|
||||||
@@ -362,7 +375,7 @@ def test_wrapper_idle_timeout_grace_accepts_disk_heartbeat(tmp_path):
|
|||||||
)
|
)
|
||||||
assert done.returncode == 0, done.stderr
|
assert done.returncode == 0, done.stderr
|
||||||
|
|
||||||
stdout, stderr = proc.communicate(timeout=90)
|
stdout, stderr = proc.communicate(timeout=150)
|
||||||
job_data = json.loads((reg_dir / f"{jid}.json").read_text())
|
job_data = json.loads((reg_dir / f"{jid}.json").read_text())
|
||||||
assert job_data.get("status") == "completed", f"status={job_data.get('status')}\n{stdout}\n{stderr}"
|
assert job_data.get("status") == "completed", f"status={job_data.get('status')}\n{stdout}\n{stderr}"
|
||||||
assert proc.returncode == 0, f"expected rc=0 while heartbeats kept landing, got {proc.returncode}\n{stdout}\n{stderr}"
|
assert proc.returncode == 0, f"expected rc=0 while heartbeats kept landing, got {proc.returncode}\n{stdout}\n{stderr}"
|
||||||
|
|||||||
Reference in new issue
Block a user