From 58b56ec94e4d387454a55ac5d0f80cf393210ac4 Mon Sep 17 00:00:00 2001 From: Jorge Miguel Silva Date: Thu, 27 Aug 2026 22:06:24 +0100 Subject: [PATCH] Stream the logs, and bound every copy of a run's output #6 delivered the logs and #8 delivered per-step progress from the same lines, but the logs were still recorded once, after the process exited. The mechanism to do better already existed: on_line delivers every line as it arrives. They are handed to the run twice a second now, so a run still going can be read as it goes. The timer is the half that matters. Checking the clock only when a line arrives looks equivalent and is not: a tool that prints "starting" and then works silently for ten minutes would show nothing for ten minutes, which is precisely the run someone wants the logs for. A thread does the waking. Only that thread delivers. 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. 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. Everything holding a run's output is now bounded by the same MAX_LOG_BYTES, including the runner's own capture, which never was. A chatty tool could otherwise decide how much memory the agent used, and nothing is lost that would have survived being recorded. TailBuffer is that ceiling, in one place rather than three: the tail is kept, not the head an error arrives at the end; a truncated beginning costs progress chatter truncation is announced a log that starts mid-sentence with no marker reads like a broken tool one line longer than the budget a tool emitting no newline -- a progress bar is truncated to its own tail redrawing with \r, or binary on stdout -- arrives as a single line of any size What a poll returns mid-run 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 costs nothing in the end. The ticker is stopped in a finally, and close() waits for a delivery already in flight: perform_run closes the log and then records the authoritative output, so a flush still running could otherwise overwrite a complete log with a partial one. Refs #6 #8 --- README.md | 19 +- main.py | 38 +++- runner.py | 26 +-- steplogs.py | 158 +++++++++++++++ tests/test_runner_seam.py | 46 +++++ tests/test_step_logs.py | 412 ++++++++++++++++++++++++++++++++++++++ 6 files changed, 672 insertions(+), 27 deletions(-) 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" + )