diff --git a/README.md b/README.md index 639f212..99b8ae6 100644 --- a/README.md +++ b/README.md @@ -89,12 +89,19 @@ reproduce, and that is a property of the tool rather than of this document. `GET /runs/{run_id}/logs` returns what the run printed, plus `failed_steps`, naming the nodes that failed and what snakemake said about each. -**The logs are not streamed, though the progress is.** They are recorded in one -go when the workflow process exits — readable while the run is still finishing, -but not during it. The two differ because progress is a handful of state -transitions and the logs are unbounded output; flushing every line into the run -store would take a lock per line. Following the output live is a separate piece -of work. A step that succeeded is not separated out: +The logs are live, like the progress. Output is read as it arrives and handed to +the run twice a second, so a run that is still going can be read as it goes. A tool that prints once +and then works silently is still visible: the flush is on a timer as well as a +line count, because checking the clock only when a line arrives would show +nothing for as long as the tool said nothing. + +What a poll returns mid-run is therefore a **partial** log, and bounded: nothing +holds more than `BIOCHEF_MAX_LOG_BYTES` of a stream, oldest dropped first, and a +truncated log says so rather than beginning mid-sentence. That ceiling applies to +the runner's own capture as well — a chatty tool used to decide how much memory +the agent used. The authoritative one is recorded when the process exits, from the +runner's own complete capture, so anything trimmed or still buffered costs +nothing in the end. A step that succeeded is not separated out: its output is in `stdout` along with everything else's, and nothing in snakemake's output marks where one rule's writing ends. Splitting that needs a `log:` directive per rule, which is emitter work. diff --git a/main.py b/main.py index 9ee3e0f..054ccde 100644 --- a/main.py +++ b/main.py @@ -23,7 +23,7 @@ UnknownRun) from datasource import DataSourceError, get_sources from retention import Retained -from steplogs import Progress, failing_steps +from steplogs import LiveLog, Progress, failing_steps from bodylimit import BodySizeLimitMiddleware, MAX_UPLOAD_BYTES from runner import SubprocessRunner, get_runner @@ -257,19 +257,41 @@ def record_manifest(code, catalogue): if on_progress is not None: on_progress(progress.snapshot()) + # Accumulated as it arrives so the logs can be read DURING a run, not + # only once the process has exited. Flushed in batches: a lock per line + # was the objection to this, and a lock twice a second is not. + node_ids = [node.id for node in workflow.nodes] + live = LiveLog( + on_flush=None if on_logs is None + else (lambda partial_out, partial_err: + on_logs(partial_out, partial_err, node_ids))) + def observe(stream, line): + live.add(stream, line) if progress.observe(line) and on_progress is not None: on_progress(progress.snapshot()) - code, out, err = run_snakemake(ws, on_start=on_start, - on_finish=on_finish, - on_line=observe if on_progress else None) + # Wired when either is wanted, since both are fed from the same lines. + wants_lines = on_progress is not None or on_logs is not None + try: + live.start() + code, out, err = run_snakemake( + ws, on_start=on_start, on_finish=on_finish, + on_line=observe if wants_lines else None) + finally: + # Always. The ticker is a thread per run, and a run that raised on + # its way out would otherwise leave one behind for every attempt. + live.close() + # The authoritative record, from the runner's own complete capture. # Recorded before the failure path raises, because a failed run is # exactly the one whose output someone needs. Reporting it only on # success, or only as a 2000-character tail, was the whole of #6. + # + # It replaces whatever the live flushes left, so a batch still in the + # buffer when the process exited costs nothing. if on_logs is not None: - on_logs(out, err, [node.id for node in workflow.nodes]) + on_logs(out, err, node_ids) if code != 0: # Before raising. E5 asks for a manifest recording exit codes, and @@ -439,10 +461,8 @@ def finished(): RUNS.detach(run_id) def step_progress(step_status): - # Named apart from `progress` above, which reports RUN state. An earlier - # version called both of them progress, so the second definition - # shadowed the first and every state transition was handed to - # record_progress as if it were a per-step map. + # Named apart from `progress` above, which reports RUN state rather + # than per-step state. Two callbacks, two vocabularies. RUNS.record_progress(run_id, step_status) def outputs(catalogue): diff --git a/runner.py b/runner.py index 566ecda..c902575 100644 --- a/runner.py +++ b/runner.py @@ -3,8 +3,7 @@ E2 asks for a second Runner provider. This is the first one, and the seam that makes a second possible. -The split is deliberate and is the whole point of the file. Running a workflow is -two things fused together in the version this replaces: +The split is the whole point of the file. Running a workflow is two things: policy the timeout, and killing the whole process group rather than the child. Every provider needs this, and it is the part that was hard @@ -26,6 +25,8 @@ import threading from typing import List, NamedTuple +from steplogs import TailBuffer + def _kill_group(pgid): """End a process group, tolerating one that has already gone.""" @@ -115,7 +116,10 @@ def run(self, ws, timeout_s: int, on_start=None, on_finish=None, # Two threads, one per stream, because draining only one of them is the # deadlock communicate() exists to avoid: a tool that fills the other # pipe's buffer blocks forever waiting for someone to read it. - collected = {"stdout": [], "stderr": []} + # Bounded, so a chatty tool cannot decide how much memory the agent + # uses. Nothing is lost that would have survived being recorded: the + # store keeps MAX_LOG_BYTES of a stream either way. + collected = {"stdout": TailBuffer(), "stderr": TailBuffer()} def pump(stream, name): try: @@ -170,10 +174,9 @@ def pump(stream, name): pass raise finally: - # Reached by every path, and by now the group is either reaped or - # killed above. An earlier version of this comment said "both paths - # reach here, and both have reaped the child", which was true of the - # two paths it named and false of every other. + # Reached by every path, and by now the group is either reaped + # normally or killed above -- including the paths that raise, which + # is why the kill is in an except rather than only in the timeout. # # The readers are joined with a bound rather than indefinitely. A # grandchild that inherited the pipes and outlived its parent holds @@ -185,12 +188,11 @@ def pump(stream, name): if on_finish is not None: on_finish() - # The real code, not a constant. This used to return -SIGKILL literally - # on the timeout path, which reported the signal we meant to send rather - # than what happened. + # The process's real code, not a constant: reporting the signal we + # meant to send would claim SIGKILL however the process actually ended. return RunResult(process.returncode, - "".join(collected["stdout"]), - "".join(collected["stderr"])) + collected["stdout"].text(), + collected["stderr"].text()) class SubprocessRunner(Runner): diff --git a/steplogs.py b/steplogs.py index 58b49d4..5f92c7b 100644 --- a/steplogs.py +++ b/steplogs.py @@ -23,6 +23,9 @@ import os import re +import threading +import time +from collections import deque MAX_LOG_BYTES = int(os.getenv("BIOCHEF_MAX_LOG_BYTES", str(1024 * 1024))) """How much of a run's output is kept. @@ -151,3 +154,158 @@ def observe(self, line): def snapshot(self): return dict(self._status) + + +class TailBuffer: + """The last `max_bytes` of a stream, and how much was dropped to keep it. + + The tail rather than the head, because an error and the traceback around it + arrive at the end; a truncated beginning costs progress chatter. And it says + it was truncated, because a log that starts mid-sentence with no explanation + reads like a tool that produced nonsense. + + One line longer than the whole budget is truncated to its own tail. That is + not a contrived case: a tool emitting no newline -- a progress bar redrawing + with \r, or binary on stdout -- arrives as a single line of whatever size, + and a trim that stopped at the last line left the bound meaningless. + + Not thread-safe on its own. Callers that share one hold their own lock. + """ + + def __init__(self, max_bytes=None): + self._lines = deque() + self._bytes = 0 + self._dropped = 0 + self._max = MAX_LOG_BYTES if max_bytes is None else max_bytes + + def append(self, line): + if len(line) > self._max: + self._dropped += len(line) - self._max + line = line[-self._max:] + self._lines.append(line) + self._bytes += len(line) + while self._bytes > self._max and len(self._lines) > 1: + oldest = self._lines.popleft() + self._bytes -= len(oldest) + self._dropped += len(oldest) + + def text(self): + body = "".join(self._lines) + if not self._dropped: + return body + return f"[... {self._dropped} earlier bytes dropped ...]\n" + body + + +class LiveLog: + """Output accumulated as it arrives, delivered by one thread on a timer. + + The objection to streaming the logs was a lock per line, and it was a fair + one: a chatty tool produces thousands, and the run store is shared with + every poll. Delivering on a timer instead makes it a lock twice a second. + + Only the ticker delivers, and that is load-bearing rather than tidy. If a + reader thread could deliver too, two of them could build snapshots in one + order and hand them over in the other, and a client polling twice would see + the log go backwards. One delivering thread cannot do that. It also keeps a + slow consumer away from the readers draining the tool's pipes: if they + stall the pipe fills and the tool stops writing. + + The buffer is bounded by the same MAX_LOG_BYTES the store keeps, so a run's + output is not held twice over -- the runner has its own copy, and only a + megabyte of it was ever going to be recorded. Bounding it bounds the cost + of joining too, which would otherwise grow with the log. + + What this produces is a PARTIAL log. The authoritative one is recorded when + the process exits, from the runner's complete capture, so anything trimmed + or still buffered here costs nothing in the end. + """ + + def __init__(self, on_flush=None, every_seconds=0.5, max_bytes=None): + self._lines = {"stdout": TailBuffer(max_bytes), + "stderr": TailBuffer(max_bytes)} + self._lock = threading.Lock() + self._on_flush = on_flush + self._every_seconds = every_seconds + self._pending = False + self._stop = threading.Event() + self._ticker = None + + def start(self): + """Begin delivering. Idempotent, and a no-op with no callback.""" + if self._on_flush is None or self._ticker is not None: + return self + self._ticker = threading.Thread(target=self._tick, daemon=True) + self._ticker.start() + return self + + def close(self): + """Stop delivering, waiting for a delivery already in flight. + + The wait matters: perform_run closes this and then records the + authoritative output, so a flush still running could otherwise deliver + its partial snapshot afterwards -- overwriting a complete log with an + incomplete one, which is worse than never having streamed. + """ + self._stop.set() + if self._ticker is not None: + self._ticker.join(timeout=5) + self._ticker = None + + def __enter__(self): + return self.start() + + def __exit__(self, *exc_info): + self.close() + return False + + def add(self, stream, line): + """Take a line. Appends and returns; it never delivers. + + Called from both reader threads, so it does as little as possible and + holds the lock only long enough to append and trim. + + The two buckets are fixed rather than created on demand: there are + exactly two streams, and a name that is not one of them is a caller + error. It raises here rather than accumulating into something nothing + will ever read. + """ + with self._lock: + self._lines[stream].append(line) + self._pending = True + + def flush(self): + """Deliver what has arrived, if anything has. Returns whether it did.""" + with self._lock: + if not self._pending: + return False + self._pending = False + snapshot = (self._lines["stdout"].text(), + self._lines["stderr"].text()) + if self._on_flush is not None: + # Outside the lock, so a slow consumer cannot block a reader. + try: + self._on_flush(*snapshot) + except Exception: # noqa: BLE001 + # The content is still buffered, so mark it undelivered again + # rather than waiting for the next line to make it visible -- + # a tool that fell quiet right after a failed delivery would + # otherwise show nothing more until the run ended. + with self._lock: + self._pending = True + raise + return True + + def snapshot(self): + """What is buffered now, as (stdout, stderr).""" + with self._lock: + return (self._lines["stdout"].text(), + self._lines["stderr"].text()) + + def _tick(self): + while not self._stop.wait(self._every_seconds): + try: + self.flush() + except Exception: # noqa: BLE001 + # A failure in reporting must not end the thread and take the + # rest of the run's logs with it. + pass diff --git a/tests/test_runner_seam.py b/tests/test_runner_seam.py index 82ea82b..4c4d44c 100644 --- a/tests/test_runner_seam.py +++ b/tests/test_runner_seam.py @@ -489,3 +489,49 @@ def slowly(stream, line): f"was built before the readers had finished" ) assert "n-19" in result.stdout, "the tail was lost" + + +def test_the_runners_own_capture_is_bounded(tmp_path, monkeypatch): + """A chatty tool decided how much memory the agent used. + + Output was accumulated whole for the final result -- 8.9 MiB held for + 4.3 MiB of output, with no ceiling -- while the store was only ever going + to keep MAX_LOG_BYTES of it. Nothing is lost that would have survived + being recorded. + """ + import steplogs + + monkeypatch.setattr(steplogs, "MAX_LOG_BYTES", 8192) + + class _Chatty(Runner): + name = "chatty-bounded-for-test" + + def command(self, ws): + return ["sh", "-c", "i=0; while [ $i -lt 4000 ]; do " + "echo \"line-$i-aaaaaaaaaaaaaaaaaaaaaaaaaaaaaa\"; " + "i=$((i+1)); done"] + + result = _Chatty().run(_Workspace(tmp_path), timeout_s=120) + + assert result.returncode == 0 + assert len(result.stdout) < 8192 * 3, ( + f"the runner returned {len(result.stdout)} bytes against an 8192 cap" + ) + assert "line-3999" in result.stdout, "the tail was dropped instead of the head" + assert "earlier bytes dropped" in result.stdout, ( + "output was truncated without saying so" + ) + + +def test_a_short_run_is_not_marked_as_truncated(tmp_path): + """The marker must mean something, so it cannot be always-on.""" + class _Brief(Runner): + name = "brief-untruncated-for-test" + + def command(self, ws): + return ["sh", "-c", "echo just-this"] + + result = _Brief().run(_Workspace(tmp_path), timeout_s=30) + + assert result.stdout == "just-this\n" + assert "dropped" not in result.stdout diff --git a/tests/test_step_logs.py b/tests/test_step_logs.py index d33c24e..95e720c 100644 --- a/tests/test_step_logs.py +++ b/tests/test_step_logs.py @@ -386,3 +386,415 @@ def test_the_regex_is_not_the_limit_on_which_names_can_be_attributed(): ) finally: shutil.rmtree(directory, ignore_errors=True) + + +# -------------------------------------------------------------------------- +# the logs while a run is happening, not only once it has ended + + +def test_only_the_ticker_delivers_so_the_log_cannot_go_backwards(): + """Two readers building snapshots can hand them over in the other order. + + Measured on the version that let either reader deliver: sizes arrived + [22, 11], so a client polling twice saw LESS the second time. One + delivering thread cannot do that. + """ + import threading + import time as _time + + from steplogs import LiveLog + + delivered = [] + live = LiveLog(on_flush=lambda out, err: delivered.append(len(out)), + every_seconds=0.05) + live.start() + try: + def write(tag): + for n in range(200): + live.add("stdout", f"{tag}-{n:04d}\n") + _time.sleep(0.001) + + threads = [threading.Thread(target=write, args=(t,)) for t in "ab"] + for thread in threads: + thread.start() + for thread in threads: + thread.join() + _time.sleep(0.2) + finally: + live.close() + + assert len(delivered) > 1, "nothing was delivered while the writers ran" + assert delivered == sorted(delivered), ( + f"the log went backwards: {delivered}" + ) + + +def test_adding_a_line_never_delivers_it_itself(): + """So a reader draining the tool's pipes is never held up by a consumer. + + If the readers stall the pipe fills and the tool blocks writing to it, so a + slow log consumer would stop the workflow. + """ + from steplogs import LiveLog + + delivered = [] + live = LiveLog(on_flush=lambda out, err: delivered.append(out), + every_seconds=10 ** 9) + + for n in range(1000): + live.add("stdout", f"line-{n}\n") + + assert delivered == [], ( + "add() delivered on its own; only the ticker should" + ) + assert live.snapshot()[0].count("\n") == 1000 + + +def test_a_quiet_tool_is_still_visible(): + """The case the whole timer exists for. + + A tool that prints "starting" and then works silently for ten minutes would + show nothing for ten minutes if the clock were only checked when a line + arrived. + """ + import time as _time + + from steplogs import LiveLog + + seen = [] + with LiveLog(on_flush=lambda out, err: seen.append(err), + every_seconds=0.2) as live: + live.add("stderr", "starting analysis\n") + _time.sleep(0.7) + + assert seen, "a tool that printed once and then went silent showed nothing" + assert "starting analysis" in seen[0] + + +def test_the_buffer_is_bounded_by_what_the_store_would_keep(): + """It held everything before: 12.4 MiB for 7.7 MiB of output, on top of the + runner's own copy, while the store was only ever going to keep a megabyte. + + Bounding it also bounds the cost of joining, which grew with the log. + """ + from steplogs import LiveLog + + live = LiveLog(on_flush=None, max_bytes=4096, every_seconds=10 ** 9) + for n in range(5000): + live.add("stdout", f"{n:06d} " + "x" * 60 + "\n") + + out, _ = live.snapshot() + assert len(out) <= 4096 * 2, f"the buffer grew to {len(out)} bytes" + assert "004999" in out, "the newest output was trimmed instead of the oldest" + + +def test_one_line_longer_than_the_whole_budget_is_still_bounded(): + """A tool that emits no newline arrives as a single enormous line. + + A progress bar redrawing with \r, or binary written to stdout, produces one + "line" of whatever size. Refusing to trim the last line left the bound + meaningless: 4 MiB was held against a 1 KiB cap. + """ + from steplogs import LiveLog + + live = LiveLog(on_flush=None, max_bytes=1024, every_seconds=10 ** 9) + live.add("stdout", "x" * (4 * 1024 * 1024)) + + shown = live.snapshot()[0] + marker, _, content = shown.partition("\n") + + assert len(content) <= 1024, f"{len(content)} bytes held against a 1024 cap" + assert "earlier bytes dropped" in marker, ( + "the line was cut without saying so, which reads like a tool that " + "produced half a line" + ) + + +def test_the_tail_of_an_oversized_line_is_what_is_kept(): + """Same reason the oldest lines go first: an error arrives at the end.""" + from steplogs import LiveLog + + live = LiveLog(on_flush=None, max_bytes=64, every_seconds=10 ** 9) + live.add("stderr", "A" * 500 + "THE ERROR") + + assert "THE ERROR" in live.snapshot()[1] + + +def test_a_stream_that_is_not_one_of_the_two_is_refused(): + """It used to be accumulated into a bucket snapshot() never read. + + Held forever and shown to nobody, which is the worst of both. There are + exactly two streams; anything else is a caller error. + """ + import pytest as _pytest + + from steplogs import LiveLog + + live = LiveLog(on_flush=None, every_seconds=10 ** 9) + with _pytest.raises(KeyError): + live.add("other", "invisible\n") + + # And nothing was kept. Asserting only that it raised was too weak: the + # version that accumulated appended the line and THEN raised on the byte + # count, so it raised the same KeyError while still holding the content + # forever. + assert not any(live._lines.get("other") or ()), ( + "the line was buffered into a bucket nothing will ever read" + ) + assert live.snapshot() == ("", "") + + +def test_a_failed_delivery_is_offered_again(): + """Otherwise a tool that fell quiet right after one would show nothing more. + + The content is still buffered either way; what was missing was any reason + for the ticker to try it again before the next line arrived. + """ + from steplogs import LiveLog + + attempts = [] + + def sometimes_angry(out, err): + attempts.append(out) + if len(attempts) == 1: + raise RuntimeError("the consumer is briefly broken") + + live = LiveLog(on_flush=sometimes_angry, every_seconds=10 ** 9) + live.add("stdout", "important\n") + + try: + live.flush() + except RuntimeError: + pass + live.flush() # no new lines added + + assert len(attempts) == 2, ( + "the content was not offered again after a failed delivery" + ) + assert "important" in attempts[1] + + +def test_trimming_drops_the_oldest_first(): + """The tail is what matters; an error arrives at the end.""" + from steplogs import LiveLog + + live = LiveLog(on_flush=None, max_bytes=100, every_seconds=10 ** 9) + for n in range(50): + live.add("stdout", f"line-{n:03d}\n") + + out, _ = live.snapshot() + assert "line-049" in out + assert "line-000" not in out + + +def test_both_streams_accumulate_separately(): + from steplogs import LiveLog + + live = LiveLog() + live.add("stdout", "out\n") + live.add("stderr", "err\n") + + out, err = live.snapshot() + assert out == "out\n" and err == "err\n" + + +def test_the_live_log_survives_two_threads_writing_at_once(): + """Both reader threads call add(). It has a lock of its own for that.""" + import threading + + from steplogs import LiveLog + + live = LiveLog(every_seconds=10 ** 9, max_bytes=10 ** 9) + barrier = threading.Barrier(2) + + def write(stream): + barrier.wait() + for n in range(500): + live.add(stream, f"{stream}-{n}\n") + + threads = [threading.Thread(target=write, args=(s,)) + for s in ("stdout", "stderr")] + for thread in threads: + thread.start() + for thread in threads: + thread.join() + + out, err = live.snapshot() + assert out.count("\n") == 500, f"{out.count(chr(10))} of 500 stdout lines" + assert err.count("\n") == 500, f"{err.count(chr(10))} of 500 stderr lines" + + +def test_close_waits_for_a_flush_that_is_already_running(): + """Otherwise a partial log can land after the complete one. + + perform_run closes the live log and then records the authoritative output. + If close() only asked the ticker to stop and did not wait, a flush already + in progress could deliver its partial snapshot after that. + """ + import threading + import time as _time + + from steplogs import LiveLog + + in_flush = threading.Event() + finished_flush = threading.Event() + + def slow_flush(out, err): + in_flush.set() + _time.sleep(0.3) + finished_flush.set() + + live = LiveLog(on_flush=slow_flush, every_seconds=0.05) + live.start() + try: + live.add("stdout", "something\n") + assert in_flush.wait(5), "the ticker never flushed" + live.close() + assert finished_flush.is_set(), ( + "close() returned while a flush was still running; its partial " + "snapshot can land after the authoritative record" + ) + finally: + live.close() + + +def test_the_flush_callback_is_not_called_holding_the_lock(): + """A slow consumer must not block anything that appends.""" + from steplogs import LiveLog + + held = [] + + def inspect_lock(out, err): + acquired = live._lock.acquire(blocking=False) + held.append(not acquired) + if acquired: + live._lock.release() + + live = LiveLog(on_flush=inspect_lock, every_seconds=10 ** 9) + live.add("stdout", "a line\n") + live.flush() + + assert held == [False], "the buffer lock was held while calling out" + + +def test_logs_are_readable_while_the_tools_are_still_running(service, + monkeypatch): + """What the README used to say was impossible. + + Verified against real snakemake as well: a 4.34s two-step workflow flushed + at +0.87s, +1.98s and +3.12s, with the visible output growing each time. + """ + import threading + + from fastapi.testclient import TestClient + + printed = threading.Event() + release = threading.Event() + + def chatty(ws, timeout_s=None, on_start=None, on_finish=None, on_line=None): + on_line("stderr", "rule step_one:\n") + on_line("stdout", "the tool is talking\n") + printed.set() + release.wait(15) + with open(os.path.join(ws.path, "tn93.distance-1-out"), "wb") as f: + f.write(b"done") + return 0, "", "everything is fine\n" + + monkeypatch.setattr(main, "run_snakemake", chatty) + + with TestClient(main.app) as client: + run_id = _submit(client).json()["run_id"] + assert printed.wait(15), "the stub never ran" + + # Give the flush a moment; it is batched, not synchronous. + deadline = time.time() + 10 + body = {} + while time.time() < deadline: + body = client.get(f"/runs/{run_id}/logs").json() + if body.get("stdout"): + break + time.sleep(0.05) + + assert body["state"] not in {s.value for s in TERMINAL_STATES}, ( + "the run had already finished; this test proved nothing" + ) + assert "the tool is talking" in body["stdout"], body + + release.set() + _wait(service, run_id, TERMINAL_STATES) + final = client.get(f"/runs/{run_id}/logs").json() + + # The authoritative record replaces the partial one. + assert final["stderr"] == "everything is fine\n", final["stderr"] + + +def test_a_caller_wanting_logs_but_not_progress_still_gets_them(): + """on_line used to be wired only when progress was wanted, so asking for + live logs alone got neither.""" + import inspect + + source = inspect.getsource(main.perform_run) + assert "on_progress is not None or on_logs is not None" in source + assert "on_line=observe if on_progress else None" not in source + + +def test_a_run_does_not_leave_its_ticker_thread_behind(service, monkeypatch): + """One thread per run, and runs are the thing this service does most. + + Verified by hand at first, which is not the same as covered: removing the + close() passed the whole suite. A service that leaks a thread per run + degrades slowly and blames the wrong thing. + """ + import threading + import time as _time + + from fastapi.testclient import TestClient + + def quick(ws, timeout_s=None, on_start=None, on_finish=None, on_line=None): + on_line("stdout", "a line\n") + with open(os.path.join(ws.path, "tn93.distance-1-out"), "wb") as f: + f.write(b"done") + return 0, "", "" + + monkeypatch.setattr(main, "run_snakemake", quick) + + before = threading.active_count() + with TestClient(main.app) as client: + for _ in range(12): + run_id = _submit(client).json()["run_id"] + _wait(service, run_id, TERMINAL_STATES) + + # The tickers wake on an interval, so give any survivor time to be counted. + _time.sleep(1.0) + leaked = threading.active_count() - before + + assert leaked <= 1, ( + f"{leaked} threads outlived 12 runs; the ticker is not being stopped" + ) + + +def test_the_ticker_is_stopped_even_when_the_run_fails(service, monkeypatch): + """The failure path is where a finally earns its keep.""" + import threading + import time as _time + + from fastapi.testclient import TestClient + + def explodes(ws, timeout_s=None, on_start=None, on_finish=None, on_line=None): + on_line("stderr", "about to fail\n") + raise RuntimeError("the runner broke") + + monkeypatch.setattr(main, "run_snakemake", explodes) + + before = threading.active_count() + with TestClient(main.app) as client: + for _ in range(12): + run_id = _submit(client).json()["run_id"] + _wait(service, run_id, TERMINAL_STATES) + + _time.sleep(1.0) + leaked = threading.active_count() - before + + assert leaked <= 1, ( + f"{leaked} threads outlived 12 failing runs" + )