|
| 1 | +"""`LiveSnapshot.size` is a cursor, and it has to describe what was read. |
| 2 | +
|
| 3 | +`frames()` in `server.live` polls the trace file's size, rebuilds a snapshot |
| 4 | +when it changes, and then stores the snapshot's own `size` as "everything up to |
| 5 | +here has been sent": |
| 6 | +
|
| 7 | + last_size = snapshot.size if snapshot.size else size |
| 8 | +
|
| 9 | +So a `size` larger than the bytes the snapshot actually covers is not a |
| 10 | +cosmetic error. It claims events that were never sent, and the next poll sees |
| 11 | +an unchanged file size and rebuilds nothing. If the writer has stopped — the |
| 12 | +run finished — the file never changes size again, the stream never rebuilds, |
| 13 | +and a finished run is streamed as a running one for as long as the page is |
| 14 | +open. |
| 15 | +
|
| 16 | +`build_snapshot` used to take that number from `path.stat()` *after* reading |
| 17 | +the events, which can exceed the bytes read two different ways: |
| 18 | +
|
| 19 | + 1. `TailRecorder` cuts at the last newline, so a half-written final line is |
| 20 | + excluded from the events but counted by `st_size`. |
| 21 | + 2. The run is appending concurrently — that is the whole premise of a live |
| 22 | + view — so bytes can land between the read and the stat. |
| 23 | +
|
| 24 | +Only (2) loses whole events, because the bytes (1) leaves out cannot yet parse |
| 25 | +into one. Both are fixed by the same change: the count comes from the read. |
| 26 | +""" |
| 27 | + |
| 28 | +from __future__ import annotations |
| 29 | + |
| 30 | +import json |
| 31 | +from pathlib import Path |
| 32 | + |
| 33 | +from grapharc.observe.trace import TailRecorder |
| 34 | +from grapharc.server.live import build_snapshot |
| 35 | + |
| 36 | + |
| 37 | +def _event( |
| 38 | + run_id: str = "r1", phase: str = "start", step: int = 1, *, terminal: bool = False |
| 39 | +) -> str: |
| 40 | + body: dict = { |
| 41 | + "ts": "2026-01-01T00:00:00+00:00", |
| 42 | + "run_id": run_id, |
| 43 | + "graph": "g", |
| 44 | + "node": "n", |
| 45 | + "phase": phase, |
| 46 | + "step": step, |
| 47 | + } |
| 48 | + if terminal: |
| 49 | + # What `compose_snapshot` actually reads to decide a run is over; a bare |
| 50 | + # `stop` phase on a node is not it. |
| 51 | + body["state_delta"] = {"termination_reason": "completed"} |
| 52 | + return json.dumps(body) |
| 53 | + |
| 54 | + |
| 55 | +def _trace(tmp_path: Path, *lines: str, tail: str = "") -> Path: |
| 56 | + path = tmp_path / "trace.jsonl" |
| 57 | + path.write_text("".join(line + "\n" for line in lines) + tail, encoding="utf-8") |
| 58 | + return path |
| 59 | + |
| 60 | + |
| 61 | +# -- what the recorder reports ---------------------------------------------- |
| 62 | + |
| 63 | + |
| 64 | +def test_the_recorder_reports_the_bytes_its_events_came_from(tmp_path): |
| 65 | + path = _trace(tmp_path, _event(), _event(step=2)) |
| 66 | + events, consumed = TailRecorder(path).read_tail() |
| 67 | + |
| 68 | + assert len(events) == 2 |
| 69 | + assert consumed == path.stat().st_size |
| 70 | + |
| 71 | + |
| 72 | +def test_a_half_written_final_line_is_excluded_from_the_count(tmp_path): |
| 73 | + """The bytes after the last newline cannot parse into an event, so counting |
| 74 | + them would claim an event that does not exist yet.""" |
| 75 | + path = _trace(tmp_path, _event(), _event(step=2), tail='{"ts": "2026-01-0') |
| 76 | + events, consumed = TailRecorder(path).read_tail() |
| 77 | + |
| 78 | + assert len(events) == 2 |
| 79 | + assert consumed < path.stat().st_size |
| 80 | + # And the count is exactly the complete prefix. |
| 81 | + assert path.read_bytes()[:consumed].endswith(b"\n") |
| 82 | + |
| 83 | + |
| 84 | +def test_a_file_with_no_complete_line_reads_as_nothing(tmp_path): |
| 85 | + path = _trace(tmp_path, tail='{"ts": "2026-01-0') |
| 86 | + |
| 87 | + assert TailRecorder(path).read_tail() == ([], 0) |
| 88 | + |
| 89 | + |
| 90 | +def test_read_events_still_returns_just_the_events(tmp_path): |
| 91 | + """The old signature is what every other caller uses.""" |
| 92 | + path = _trace(tmp_path, _event(), _event(step=2)) |
| 93 | + |
| 94 | + assert TailRecorder(path).read_events() == TailRecorder(path).read_tail()[0] |
| 95 | + |
| 96 | + |
| 97 | +# -- the cursor the stream stores -------------------------------------------- |
| 98 | + |
| 99 | + |
| 100 | +def test_the_snapshot_size_never_exceeds_what_it_read(tmp_path, monkeypatch): |
| 101 | + """The wedge, demonstrated. |
| 102 | +
|
| 103 | + A writer appending between the read and the stat is the ordinary case for a |
| 104 | + live view, not an exotic one; it is made deterministic here by appending |
| 105 | + from inside the read. With `size` taken from a later `stat()`, the snapshot |
| 106 | + reported 2 events and a cursor past the third, and `frames()` would store |
| 107 | + that cursor and skip the third event for as long as the file stayed that |
| 108 | + size. |
| 109 | + """ |
| 110 | + path = _trace(tmp_path, _event(), _event(step=2)) |
| 111 | + real_read_tail = TailRecorder.read_tail |
| 112 | + |
| 113 | + def read_then_append(self, run_id=None): |
| 114 | + result = real_read_tail(self, run_id) |
| 115 | + # The writer lands in the window. |
| 116 | + with open(self.path, "a", encoding="utf-8") as handle: |
| 117 | + handle.write(_event(phase="stop", step=3) + "\n") |
| 118 | + return result |
| 119 | + |
| 120 | + monkeypatch.setattr(TailRecorder, "read_tail", read_then_append) |
| 121 | + snapshot = build_snapshot(tmp_path, "trace.jsonl", None) |
| 122 | + |
| 123 | + assert len(snapshot.run_ids) == 1 |
| 124 | + # The cursor describes the two events that were read, not the three bytes' |
| 125 | + # worth now on disk. |
| 126 | + assert snapshot.size < path.stat().st_size, ( |
| 127 | + "size claims bytes the snapshot never read; frames() will skip them" |
| 128 | + ) |
| 129 | + |
| 130 | + |
| 131 | +def test_the_skipped_event_is_picked_up_on_the_next_read(tmp_path, monkeypatch): |
| 132 | + """The point of the fix: the stream self-corrects. |
| 133 | +
|
| 134 | + Because the cursor stopped short, the file's size now differs from it, so |
| 135 | + the next poll rebuilds and the event arrives. That is the difference |
| 136 | + between a one-poll delay and a stream that never recovers. |
| 137 | + """ |
| 138 | + path = _trace(tmp_path, _event(), _event(step=2)) |
| 139 | + real_read_tail = TailRecorder.read_tail |
| 140 | + appended = {"done": False} |
| 141 | + |
| 142 | + def read_then_append_once(self, run_id=None): |
| 143 | + result = real_read_tail(self, run_id) |
| 144 | + if not appended["done"]: |
| 145 | + appended["done"] = True |
| 146 | + with open(self.path, "a", encoding="utf-8") as handle: |
| 147 | + handle.write(_event(phase="stop", step=3) + "\n") |
| 148 | + return result |
| 149 | + |
| 150 | + monkeypatch.setattr(TailRecorder, "read_tail", read_then_append_once) |
| 151 | + first = build_snapshot(tmp_path, "trace.jsonl", None) |
| 152 | + |
| 153 | + monkeypatch.setattr(TailRecorder, "read_tail", real_read_tail) |
| 154 | + second = build_snapshot(tmp_path, "trace.jsonl", None) |
| 155 | + |
| 156 | + assert first.size < path.stat().st_size # cursor short, so the poll refires |
| 157 | + assert second.size == path.stat().st_size |
| 158 | + assert second.size > first.size |
| 159 | + |
| 160 | + |
| 161 | +def test_a_quiet_complete_trace_reports_the_whole_file(tmp_path): |
| 162 | + """The ordinary case must not regress: nothing is being appended, the last |
| 163 | + line is complete, so the cursor is the file size and the stream settles.""" |
| 164 | + path = _trace(tmp_path, _event(), _event(step=2), _event(phase="stop", step=3)) |
| 165 | + snapshot = build_snapshot(tmp_path, "trace.jsonl", None) |
| 166 | + |
| 167 | + assert snapshot.size == path.stat().st_size |
| 168 | + |
| 169 | + |
| 170 | +def test_an_empty_or_missing_trace_is_still_a_waiting_snapshot(tmp_path): |
| 171 | + """The URL is posted before the run starts writing.""" |
| 172 | + (tmp_path / "trace.jsonl").write_text("", encoding="utf-8") |
| 173 | + |
| 174 | + assert build_snapshot(tmp_path, "trace.jsonl", None).size == 0 |
| 175 | + |
| 176 | + |
| 177 | +def test_a_finished_run_stops_reporting_itself_as_running(tmp_path, monkeypatch): |
| 178 | + """The symptom a viewer actually sees, end to end. |
| 179 | +
|
| 180 | + Reproduced against the old code: an append landing in the window left the |
| 181 | + cursor equal to the file's final size while `done` was still False, so |
| 182 | + `frames()` had no reason to rebuild and never learned the run had stopped. |
| 183 | + The page showed a finished run as running for as long as it stayed open. |
| 184 | +
|
| 185 | + With the cursor taken from the read, it falls short of the file, the next |
| 186 | + poll rebuilds, and that rebuild sees the `stop` event. |
| 187 | + """ |
| 188 | + path = _trace(tmp_path, _event(), _event(step=2)) |
| 189 | + real_read_tail = TailRecorder.read_tail |
| 190 | + fired = {"n": 0} |
| 191 | + |
| 192 | + def read_then_append_once(self, run_id=None): |
| 193 | + result = real_read_tail(self, run_id) |
| 194 | + fired["n"] += 1 |
| 195 | + if fired["n"] == 1: |
| 196 | + with open(self.path, "a", encoding="utf-8") as handle: |
| 197 | + handle.write(_event(phase="stop", step=3, terminal=True) + "\n") |
| 198 | + return result |
| 199 | + |
| 200 | + monkeypatch.setattr(TailRecorder, "read_tail", read_then_append_once) |
| 201 | + first = build_snapshot(tmp_path, "trace.jsonl", None) |
| 202 | + final_size = path.stat().st_size |
| 203 | + |
| 204 | + # The old failure was exactly this pair: cursor == final size, done False. |
| 205 | + assert not (first.size == final_size and not first.done), ( |
| 206 | + "cursor matches the finished file while done is False: frames() will " |
| 207 | + "never rebuild and the run streams as running forever" |
| 208 | + ) |
| 209 | + |
| 210 | + monkeypatch.setattr(TailRecorder, "read_tail", real_read_tail) |
| 211 | + second = build_snapshot(tmp_path, "trace.jsonl", None) |
| 212 | + |
| 213 | + assert second.done is True |
| 214 | + assert second.size == final_size |
0 commit comments