Stream the logs, and bound every copy of a run's output

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.

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):

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):

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

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