From 38c3df482ef39e08c33e1c19fbf57536751be91f Mon Sep 17 00:00:00 2001 From: Sean Ahern Date: Sat, 19 Sep 2026 01:23:34 -0400 Subject: [PATCH 1/6] Name the worker-status bucket for branches that have none branch_worker_status is NULL for a branch that is done or unqueued, and _collection_summary used that NULL as a dict key. The server encodes with sort_keys, which cannot order None against the string keys beside it, so one finished branch anywhere in scope made the entire queue report raise TypeError rather than merely carry an oddly named bucket. Against the production queue that is every request: 54,185 of 54,206 branches are done, and the Queue view cannot be served at all. The bucket is named not_applicable, matching cache_state's spelling for the same idea -- a row for which the dimension does not apply. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_01PpacbZRqT46b4Duhxq5QNQ --- report_model.py | 48 ++++++++++++++++++++++++++++++++++++++ tests/test_report_model.py | 20 ++++++++++++++++ 2 files changed, 68 insertions(+) diff --git a/report_model.py b/report_model.py index a644b443..8c95cb1a 100644 --- a/report_model.py +++ b/report_model.py @@ -2759,12 +2759,22 @@ def _row_matches_branch_target(row, branch_target, prefix): return False +#: The worker-status bucket for branches that have no worker dimension at all. +#: `branch_worker_status` is NULL for a branch that is done or unqueued, and a +#: None key cannot be ordered against the string keys beside it -- which makes +#: the whole report unserializable under `sort_keys`, not merely oddly named. +#: The spelling matches `cache_state`'s "not_applicable" for the same idea. +NO_WORKER_STATUS_BUCKET = "not_applicable" + + def _collection_summary(rows): by_status = {} by_worker_status = {} for row in rows: status = row["branch_status"] worker_status = row["branch_worker_status"] + if worker_status is None: + worker_status = NO_WORKER_STATUS_BUCKET by_status[status] = by_status.get(status, 0) + 1 by_worker_status[worker_status] = by_worker_status.get(worker_status, 0) + 1 return { @@ -3906,6 +3916,44 @@ def _opener_erd_summaries(sources, openers, report, cache, return summaries +def opener_completion_signal(sources): + """A token that changes when an opener's tree finishes, and not otherwise. + + The leaderboard's answer is the set of openers whose whole tree is solved. + That set grows when an opener's work completes — about every 27 minutes on + the production swarm — against a client that asks every two seconds. Two + indexed counts over `opener_work` say whether it could have grown, for + about a tenth of a millisecond. + + **The obvious signal does not work, and the reason is worth keeping.** A + watermark over branch results (`MAX(updated_at)`) is the conservative + choice and is useless here: it covers every branch written at any depth, + and those land every few seconds — measured at 189 changes in 180 seconds, + a median of one per second, against a leaderboard answer that moves once + per 1,650. A signal that fires faster than the poll interval caches + nothing. + + This signal is a hint, never an answer: the caller still rescreens every + opener against the cache. It is also not exhaustive — a repair or an + import changes the cache without touching the queue — so a caller must + pair it with a staleness bound. Returns None when the queue cannot be + read, which callers must treat as "assume changed". + """ + try: + connection = sqlite3.connect( + f"file:{sources.queue_path}?mode=ro", uri=True) + except (sqlite3.Error, OSError): + return None + try: + return tuple(connection.execute( + "SELECT COUNT(*), MAX(opener_work_id) FROM opener_work " + "WHERE state = 'complete'").fetchone()) + except sqlite3.Error: + return None + finally: + connection.close() + + def _score_cache_file_signature(cache_path): """Identify the cache generation, including uncheckpointed WAL writes.""" def file_signature(path): diff --git a/tests/test_report_model.py b/tests/test_report_model.py index f87181bb..318805c5 100644 --- a/tests/test_report_model.py +++ b/tests/test_report_model.py @@ -15,6 +15,8 @@ import erd_search import report_model from report_model import ( + _collection_summary, + NO_WORKER_STATUS_BUCKET, ReportFilters, ReportRequest, ReportOpeners, @@ -1856,6 +1858,24 @@ def _leaderboard_sources(self, answers, candidates): telemetry_path=self.telemetry_path, ) + def test_a_summary_of_branches_with_no_worker_status_stays_serializable(self): + # branch_worker_status is NULL for a done or unqueued branch, and a + # None key cannot be ordered against the string keys beside it. The + # server encodes with sort_keys, so one such branch made the whole + # queue report unserializable rather than merely oddly labelled -- and + # a queue holding any finished branch has one. + summary = _collection_summary([ + {"branch_status": "evaluating", "branch_worker_status": "active"}, + {"branch_status": "done", "branch_worker_status": None}, + {"branch_status": "queued", "branch_worker_status": "waiting"}, + ]) + self.assertEqual( + summary["branch_count_by_worker_status"], + {"active": 1, NO_WORKER_STATUS_BUCKET: 1, "waiting": 1}, + ) + self.assertNotIn(None, summary["branch_count_by_worker_status"]) + json.dumps(summary, sort_keys=True) + def test_leaderboard_ranks_complete_openers_by_erd(self): # With two answers, an opener that separates them into singletons is # complete with no cache reads at all: crane/slate are also answers From ad7ee490e3f4656bb2ce6ce1651230ceecc365e5 Mon Sep 17 00:00:00 2001 From: Sean Ahern Date: Sat, 19 Sep 2026 01:23:47 -0400 Subject: [PATCH 2/6] Rebuild the leaderboard on opener completion, not on every poll The client polls every two seconds and the leaderboard is rebuilt for each poll, so the server is never idle and requests queue behind one another -- which is what the shared-collection lock was papering over rather than fixing. The answer only grows when an opener's tree finishes, which the production swarm does about every 27 minutes. opener_completion_signal reads that in two indexed counts over opener_work (~0.14ms), and report_server serves the cached report while it stands still. Measured end to end against the production cache: a repeat leaderboard request went from 18.4s to 0.005s, serving identical bytes. The encoded body is cached with the report, because re-encoding 3.7MB per poll is its own cost once the build is gone. The obvious signal does not work, and the measurement is in AGENTS.md so it is not retried: a branch-result watermark (MAX(updated_at)) cannot miss a change, but it covers branches written at every depth and moved 189 times in 180 seconds against an answer that moves once per 1,650. A 90-second sample during a quiet stretch shows zero changes and reads as a green light, which is how this was nearly shipped. The signal is a hint, never an answer: the build still rescreens every opener, so a stale signal costs freshness and not correctness. It is also not exhaustive -- a repair or an import changes the cache without completing queue work -- so REPORT_CACHE_MAX_AGE_SECONDS bounds how long that can go unnoticed, and an unreadable signal rebuilds rather than asserting a freshness the server cannot support. Queue-backed reports are deliberately excluded. Their subject is what the swarm is doing now, so serving one a minute old would make a liveness dashboard report a liveness it no longer has. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_01PpacbZRqT46b4Duhxq5QNQ --- AGENTS.md | 43 +++++++++++++++++ report_server.py | 96 +++++++++++++++++++++++++++++-------- tests/test_report_server.py | 87 +++++++++++++++++++++++++++++++++ 3 files changed, 205 insertions(+), 21 deletions(-) diff --git a/AGENTS.md b/AGENTS.md index f6e7582d..95b4c0fe 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -510,6 +510,49 @@ winner is decided once, at finalize, and is recorded per branch in `branch_finalize_log.hint_was_winner` — never added to a process counter that several workers each contributed placements to. +### An expensive report is rebuilt on its own event, not on the poll + +`report_client.html` polls every two seconds. A leaderboard build rescreens the +whole opener vocabulary, so serving one per poll means the server is never idle +and requests queue behind each other — which is what +`report_server.collect_report_once`'s shared-collection lock was papering over. + +`REVALIDATED_REPORT_KINDS` names the reports served from a cache that +revalidates instead of rebuilding. A kind belongs there only if its answer is a +function of something the signal covers. **The queue-backed reports are +deliberately excluded**: their subject is what the swarm is doing right now, so +serving one a minute old would make a liveness dashboard report a liveness it +no longer has. Caching them would be cheap and wrong. + +`report_model.opener_completion_signal` is the signal: two indexed counts over +`opener_work` where `state = 'complete'`, about 0.14 ms. It moves exactly when +an opener's tree finishes, which is what grows the leaderboard's answer. + +**The obvious signal is a branch-result watermark, and it does not work.** +`MAX(updated_at)` over the branch tables is the conservative choice — it cannot +miss a change — and it is useless here, because it covers every branch written +at any depth. Measured on the production swarm: the watermark changed **189 +times in 180 seconds** (median gap 1.0 s) while openers completed about **once +every 27 minutes**. A signal that fires faster than the poll interval caches +nothing. Do not reach for the watermark again without re-measuring that ratio; +a 90-second sample taken during a quiet stretch shows zero changes and reads as +a green light. + +The signal is a hint and never an answer: the build still rescreens every +opener against the cache, so a stale signal costs freshness and never +correctness. It is also not exhaustive — a repair, a reverification or an +import changes the cache without completing any queue work — so +`REPORT_CACHE_MAX_AGE_SECONDS` bounds how long such a change can go unnoticed. +That age is a backstop for the rare case, not the mechanism. A signal that +cannot be read at all returns `None`, which must be treated as "assume +changed": serving a cached report on no information asserts a freshness the +server cannot support. + +The encoded body is cached alongside the report, because re-encoding a +multi-megabyte ranking on every poll is its own cost once the build is gone. +Measured end to end on the production cache: a repeat leaderboard request went +from 18.4 s to 0.005 s, serving identical bytes. + ### Completed work has two records, and they can disagree "Already solved" is answered by the **permanent cache** diff --git a/report_server.py b/report_server.py index 2e3144b7..64f943d5 100644 --- a/report_server.py +++ b/report_server.py @@ -11,6 +11,7 @@ import os import re import sys +import time from threading import Event, Lock from urllib.parse import parse_qs, urlsplit @@ -24,6 +25,7 @@ GROUP_BY_STRATEGIES, SCHEMA_VERSION, OPENER_GROUP_BY_STRATEGIES, + opener_completion_signal, OPENER_SORT_FIELDS, OPENER_STATES, TREE_CURSOR_PATTERN, @@ -85,15 +87,31 @@ class InvalidRequest(ValueError): pass -class InFlightLeaderboardReport: - """One leaderboard collection shared by concurrent identical requests.""" +class InFlightReport: + """One report collection shared by concurrent identical requests.""" def __init__(self): self.completed = Event() - self.report = None + self.body = None self.error = None +#: Report kinds served from the revalidating cache. A kind belongs here only +#: if its answer is a function of the branch results `branch_result_watermark` +#: covers. The queue-backed reports are deliberately absent: their subject is +#: what the swarm is doing right now, so serving one a few seconds old would +#: make a liveness dashboard report a liveness it no longer has. +REVALIDATED_REPORT_KINDS = frozenset({"leaderboard"}) + +#: How long a cached report may be served while its signal stands still. The +#: signal names openers whose queue work finished, so a change reaching the +#: cache by another route — a repair, a reverification, an import — is +#: invisible to it; this bounds how long such a change can go unnoticed. It is +#: a backstop for the rare case, not the mechanism: an opener completing is +#: caught within one poll. +REPORT_CACHE_MAX_AGE_SECONDS = 120 + + @dataclass(frozen=True) class ServerConfiguration: sources: ReportOpeners @@ -387,32 +405,67 @@ def load_fixtures(directory): def make_handler(configuration): - leaderboard_reports = {} - leaderboard_reports_lock = Lock() - - def collect_leaderboard_once(request): - with leaderboard_reports_lock: - in_flight = leaderboard_reports.get(request) + in_flight_reports = {} + in_flight_lock = Lock() + cached_reports = {} + cached_reports_lock = Lock() + + def encode_report(report): + return json.dumps(report, sort_keys=True).encode("utf-8") + + def cached_report_body(request): + """Serve a revalidated report kind, rebuilding only when it could differ. + + Rebuilding the leaderboard means rescreening the whole opener + vocabulary; the signal that says whether the answer could have moved is + two indexed counts. Asking the cheap question on every request and the + expensive one only when the answer changes is the whole point: an + opener completes about every 27 minutes against a client that polls + every two seconds. + + The encoded body is cached with the report, because re-encoding a + multi-megabyte ranking on every poll is its own cost once the build is + gone. + """ + token = opener_completion_signal(configuration.sources) + now = time.time() + if token is not None: + with cached_reports_lock: + entry = cached_reports.get(request) + if (entry is not None and entry[0] == token + and now - entry[1] <= REPORT_CACHE_MAX_AGE_SECONDS): + return entry[2] + body = collect_report_once(request) + if token is not None: + with cached_reports_lock: + cached_reports[request] = (token, now, body) + return body + + def collect_report_once(request): + """Collect `request` once, however many callers are waiting on it.""" + with in_flight_lock: + in_flight = in_flight_reports.get(request) if in_flight is None: - in_flight = InFlightLeaderboardReport() - leaderboard_reports[request] = in_flight + in_flight = InFlightReport() + in_flight_reports[request] = in_flight is_builder = True else: is_builder = False if is_builder: try: - in_flight.report = collect_report(configuration.sources, request) + in_flight.body = encode_report( + collect_report(configuration.sources, request)) except Exception as error: in_flight.error = error finally: in_flight.completed.set() - with leaderboard_reports_lock: - leaderboard_reports.pop(request, None) + with in_flight_lock: + in_flight_reports.pop(request, None) else: in_flight.completed.wait() if in_flight.error is not None: raise in_flight.error - return in_flight.report + return in_flight.body class ReportHandler(BaseHTTPRequestHandler): def log_message(self, _format, *_args): @@ -471,13 +524,14 @@ def do_GET(self): return try: if configuration.fixtures is not None: - report = configuration.fixtures[ + body = encode_report(configuration.fixtures[ fixture_name_for_request(target.path, request) - ] - elif request.report_kind == "leaderboard": - report = collect_leaderboard_once(request) + ]) + elif request.report_kind in REVALIDATED_REPORT_KINDS: + body = cached_report_body(request) else: - report = collect_report(configuration.sources, request) + body = encode_report( + collect_report(configuration.sources, request)) except ValueError as error: if hasattr(error, "candidates"): report = collect_ambiguous_branch_reference_report( @@ -496,7 +550,7 @@ def do_GET(self): print("report server: report collection failed", file=sys.stderr) self._error(500, "server_error", "report collection failed") return - self._json(200, report) + self._write(200, "application/json; charset=utf-8", body) def _method_not_allowed(self): self._error( diff --git a/tests/test_report_server.py b/tests/test_report_server.py index 27eeefbf..add37921 100644 --- a/tests/test_report_server.py +++ b/tests/test_report_server.py @@ -566,6 +566,93 @@ def collect_slowly(_sources, _request): ) +class RevalidatedReportCacheTest(ReportServerTest): + """When a rebuilt leaderboard is served, and when a cached one is.""" + + def setUp(self): + super().setUp() + self.report = load_fixtures(FIXTURE_DIRECTORY)["leaderboard.json"] + self.calls = 0 + + def collect_counting(self, _sources, _request): + self.calls += 1 + return self.report + + def get_leaderboard(self, base_url, times=1): + for _ in range(times): + status, _headers, body = request(base_url, "/api/view/leaderboard") + self.assertEqual(status, 200) + return body + + def test_an_unchanged_completion_signal_serves_the_cached_report(self): + # The signal is what the whole cache turns on: while no opener has + # finished, a rebuild cannot produce a different ranking, so asking for + # one is pure cost against a client that polls every two seconds. + with patch("report_server.collect_report", side_effect=self.collect_counting), \ + patch("report_server.opener_completion_signal", return_value=(3, 3)): + with running_server(self.live_configuration) as base_url: + first = self.get_leaderboard(base_url) + self.get_leaderboard(base_url, times=4) + last = self.get_leaderboard(base_url) + self.assertEqual(self.calls, 1) + self.assertEqual(first, last) + + def test_a_moved_completion_signal_rebuilds(self): + signals = iter([(3, 3), (3, 3), (4, 4), (4, 4)]) + with patch("report_server.collect_report", side_effect=self.collect_counting), \ + patch("report_server.opener_completion_signal", + side_effect=lambda _sources: next(signals)): + with running_server(self.live_configuration) as base_url: + self.get_leaderboard(base_url, times=4) + self.assertEqual(self.calls, 2) + + def test_an_unreadable_signal_rebuilds_every_time(self): + # None means the queue could not be read, so nothing is known about + # whether an opener finished. Serving a cached report on no + # information would be asserting freshness the server cannot support. + with patch("report_server.collect_report", side_effect=self.collect_counting), \ + patch("report_server.opener_completion_signal", return_value=None): + with running_server(self.live_configuration) as base_url: + self.get_leaderboard(base_url, times=3) + self.assertEqual(self.calls, 3) + + def test_a_cached_report_expires_even_though_the_signal_stands_still(self): + # A repair or an import changes the cache without completing any opener + # work, so the signal cannot see it. The age bound is what keeps such a + # change from going unnoticed indefinitely. + # The server's own machinery reads the clock too, so the fake advances + # on the cache's reads rather than replacing time everywhere. + now = [1000.0] + + def collect_and_age(sources, request): + now[0] += report_server.REPORT_CACHE_MAX_AGE_SECONDS + 1 + return self.collect_counting(sources, request) + + with patch("report_server.collect_report", side_effect=collect_and_age), \ + patch("report_server.opener_completion_signal", return_value=(3, 3)), \ + patch.object(report_server.time, "time", lambda: now[0]): + with running_server(self.live_configuration) as base_url: + self.get_leaderboard(base_url, times=2) + self.assertEqual(self.calls, 2) + + def test_a_queue_backed_report_is_never_served_from_the_cache(self): + # The queue reports exist to say what the swarm is doing now. Serving + # one a minute old would make a liveness dashboard report a liveness it + # no longer has, so they are collected on every request however cheap + # caching them would be. + self.assertNotIn("queue", report_server.REVALIDATED_REPORT_KINDS) + queue_report = load_fixtures(FIXTURE_DIRECTORY)["queue.json"] + with patch("report_server.collect_report", + side_effect=lambda _s, _r: ( + setattr(self, "calls", self.calls + 1) or queue_report)), \ + patch("report_server.opener_completion_signal", return_value=(3, 3)): + with running_server(self.live_configuration) as base_url: + for _ in range(3): + status, _headers, _body = request(base_url, "/api/view/queue") + self.assertEqual(status, 200) + self.assertEqual(self.calls, 3) + + class ReportServerMainTest(unittest.TestCase): def test_port_collision_exits_cleanly_instead_of_crashing(self): with running_server(fixture_configuration()) as base_url: From 9e1e89bee6774e29688db5f9287aed4b009188ef Mon Sep 17 00:00:00 2001 From: Sean Ahern Date: Sat, 19 Sep 2026 01:25:55 -0400 Subject: [PATCH 3/6] Name the signal the revalidated kinds are chosen against The comment still pointed at a branch-result watermark, which is not what decides the set -- opener_completion_signal is. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_01PpacbZRqT46b4Duhxq5QNQ --- report_server.py | 9 +++++---- 1 file changed, 5 insertions(+), 4 deletions(-) diff --git a/report_server.py b/report_server.py index 64f943d5..559e8d83 100644 --- a/report_server.py +++ b/report_server.py @@ -97,10 +97,11 @@ def __init__(self): #: Report kinds served from the revalidating cache. A kind belongs here only -#: if its answer is a function of the branch results `branch_result_watermark` -#: covers. The queue-backed reports are deliberately absent: their subject is -#: what the swarm is doing right now, so serving one a few seconds old would -#: make a liveness dashboard report a liveness it no longer has. +#: if its answer is a function of what `opener_completion_signal` covers — the +#: set of openers whose tree is finished. The queue-backed reports are +#: deliberately absent: their subject is what the swarm is doing right now, so +#: serving one a few seconds old would make a liveness dashboard report a +#: liveness it no longer has. REVALIDATED_REPORT_KINDS = frozenset({"leaderboard"}) #: How long a cached report may be served while its signal stands still. The From 5d5f8e2f90307da9e38e9141534f203e2d2deee7 Mon Sep 17 00:00:00 2001 From: Sean Ahern Date: Sat, 19 Sep 2026 16:34:05 -0400 Subject: [PATCH 4/6] Cache a leaderboard body only when its builder can vouch for it Five review findings, each with a test that fails without its fix. A waiter must not file the builder's body under its own token. A waiter joins a build that started before the waiter read the signal, so the body it receives can predate the token it holds; filing under that token publishes a body older than the state the token names, and every later poll matching it is served the stale ranking until the entry expires. Only the builder files now, under the token it read before starting -- conservative in the safe direction, because a signal that moves mid-build leaves the entry under the older token and the next request rebuilds. A build that recorded a source error is not cached. collect_leaderboard_report catches its own SQLite and OS errors and returns an ordinary report with an empty ranking and the error on the source, so a degraded build is a normal-looking 200. Caching one pinned an empty leaderboard for up to the staleness bound, where every poll used to recover on the next request. A source that was never consulted carries ok=false with no error, which is not a failure and must not read as one. The cache is bounded. Its key is the request, and a request carries a user-controlled limit, so distinct ?limit= values each pinned a multi-megabyte body for the life of the process; expiry alone only stops an entry being reused, never removes it. Expired entries are purged on write and the oldest go after that. An entry is aged from when its build started, not when it finished: the age is meant to say how stale the content is, and the content is as old as the read that produced it. opener_completion_signal builds its URI through erd_queue.read_only_database_uri rather than interpolating the path. A queue path holding '?' or '#' is a valid filename and a URI delimiter, so interpolation opened a different database, returned None, and silently disabled revalidation. Its two failure arms now have tests that call it rather than mock it. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_01PpacbZRqT46b4Duhxq5QNQ --- report_model.py | 3 +- report_server.py | 88 +++++++++++++++++++++++++++++----- tests/test_report_model.py | 61 ++++++++++++++++++++++++ tests/test_report_server.py | 95 +++++++++++++++++++++++++++++++++++++ 4 files changed, 234 insertions(+), 13 deletions(-) diff --git a/report_model.py b/report_model.py index 8c95cb1a..f163d7d8 100644 --- a/report_model.py +++ b/report_model.py @@ -21,6 +21,7 @@ best_first_rank, derive_telemetry_path, disk_stats, + read_only_database_uri, ) from runtime_paths import ( DEFAULT_ANSWER_LIST_PATH, @@ -3941,7 +3942,7 @@ def opener_completion_signal(sources): """ try: connection = sqlite3.connect( - f"file:{sources.queue_path}?mode=ro", uri=True) + read_only_database_uri(sources.queue_path), uri=True) except (sqlite3.Error, OSError): return None try: diff --git a/report_server.py b/report_server.py index 559e8d83..a3e0aa03 100644 --- a/report_server.py +++ b/report_server.py @@ -94,6 +94,9 @@ def __init__(self): self.completed = Event() self.body = None self.error = None + # False when the build recorded an error on one of its sources, which + # `collect_leaderboard_report` does instead of raising. + self.intact = False #: Report kinds served from the revalidating cache. A kind belongs here only @@ -112,6 +115,12 @@ def __init__(self): #: caught within one poll. REPORT_CACHE_MAX_AGE_SECONDS = 120 +#: How many distinct revalidated bodies to hold. The cache key is the request, +#: and a request carries a user-controlled `limit`, so distinct `?limit=` values +#: would otherwise each pin a multi-megabyte body for the life of the process. +#: A handful covers the limits one session actually uses. +REPORT_CACHE_MAX_ENTRIES = 8 + @dataclass(frozen=True) class ServerConfiguration: @@ -414,6 +423,49 @@ def make_handler(configuration): def encode_report(report): return json.dumps(report, sort_keys=True).encode("utf-8") + def report_is_intact(report): + """True when no source the report consulted reported an error. + + `collect_leaderboard_report` catches its own SQLite and OS errors and + returns an ordinary report with an empty ranking and the error recorded + on the source, so a degraded build is a normal-looking 200. Caching + one would pin an empty leaderboard for as long as the entry lives, + where before it was rebuilt on the next poll. A source that was simply + not consulted carries `ok: false` with no error, which is not a + failure. + """ + return not any( + source.get("error") + for source in (report.get("sources") or {}).values() + if isinstance(source, dict) + ) + + def store_cached_body(request, token, started_at, body): + """Record a body under the token its build read, aged from its start. + + `started_at` is when the build began, not when it finished: the entry's + age is meant to say how stale its *content* is, and the content is as + old as the read that produced it. Timing from completion would let a + slow build's entry outlive its own staleness bound. + + Entries are bounded because the request is the key and a request + carries a user-controlled `limit`: every distinct `?limit=` value is a + separate multi-megabyte body, and expiry alone only stops an entry + being reused, never removes it. Expired entries go first, and the + oldest go after that if the cap is still exceeded. + """ + now = time.time() + with cached_reports_lock: + cached_reports[request] = (token, started_at, body) + for stale in [ + key for key, entry in cached_reports.items() + if now - entry[1] > REPORT_CACHE_MAX_AGE_SECONDS + ]: + del cached_reports[stale] + while len(cached_reports) > REPORT_CACHE_MAX_ENTRIES: + oldest = min(cached_reports, key=lambda key: cached_reports[key][1]) + del cached_reports[oldest] + def cached_report_body(request): """Serve a revalidated report kind, rebuilding only when it could differ. @@ -429,21 +481,30 @@ def cached_report_body(request): gone. """ token = opener_completion_signal(configuration.sources) - now = time.time() + started_at = time.time() if token is not None: with cached_reports_lock: entry = cached_reports.get(request) if (entry is not None and entry[0] == token - and now - entry[1] <= REPORT_CACHE_MAX_AGE_SECONDS): + and time.time() - entry[1] <= REPORT_CACHE_MAX_AGE_SECONDS): return entry[2] - body = collect_report_once(request) - if token is not None: - with cached_reports_lock: - cached_reports[request] = (token, now, body) - return body - - def collect_report_once(request): - """Collect `request` once, however many callers are waiting on it.""" + return collect_report_once(request, token, started_at) + + def collect_report_once(request, token, started_at): + """Collect `request` once, however many callers are waiting on it. + + **Only the caller that builds files the result**, under the token it + read before starting. A waiter receives a body built by someone else, + from a read of the cache that may predate the waiter's own token — so a + waiter filing under its own token would record a body older than the + state that token names, and every later poll matching that token would + be served it. That inverts the one guarantee the design rests on: a + stale signal must cost freshness, never correctness. + + Filing under the builder's token is conservative in the safe + direction. If the signal moves mid-build, the entry is filed under the + older token, the next request misses, and the report is rebuilt. + """ with in_flight_lock: in_flight = in_flight_reports.get(request) if in_flight is None: @@ -454,8 +515,9 @@ def collect_report_once(request): is_builder = False if is_builder: try: - in_flight.body = encode_report( - collect_report(configuration.sources, request)) + report = collect_report(configuration.sources, request) + in_flight.body = encode_report(report) + in_flight.intact = report_is_intact(report) except Exception as error: in_flight.error = error finally: @@ -466,6 +528,8 @@ def collect_report_once(request): in_flight.completed.wait() if in_flight.error is not None: raise in_flight.error + if is_builder and token is not None and in_flight.intact: + store_cached_body(request, token, started_at, in_flight.body) return in_flight.body class ReportHandler(BaseHTTPRequestHandler): diff --git a/tests/test_report_model.py b/tests/test_report_model.py index 318805c5..141966a8 100644 --- a/tests/test_report_model.py +++ b/tests/test_report_model.py @@ -15,6 +15,7 @@ import erd_search import report_model from report_model import ( + opener_completion_signal, _collection_summary, NO_WORKER_STATUS_BUCKET, ReportFilters, @@ -1858,6 +1859,66 @@ def _leaderboard_sources(self, answers, candidates): telemetry_path=self.telemetry_path, ) + def _signal_sources(self, queue_path): + directory = self.temporary_directory.name + return ReportOpeners( + queue_path, + os.path.join(directory, "signal_cache.sqlite3"), + os.path.join(directory, "signal_answers.txt"), + os.path.join(directory, "signal_candidates.txt"), + os.path.join(directory, "signal_telemetry.sqlite3"), + ) + + def test_opener_completion_signal_counts_only_finished_opener_work(self): + directory = self.temporary_directory.name + queue_path = os.path.join(directory, "signal_queue.sqlite3") + queue = ERDQueue(queue_path) + queue._conn.execute( + "INSERT INTO opener_work (opener, requested_priority, requested_at," + " state) VALUES ('crane', 10, 1, 'queued')") + queue.close() + sources = self._signal_sources(queue_path) + self.assertEqual(opener_completion_signal(sources), (0, None)) + + queue = ERDQueue(queue_path) + queue._conn.execute( + "INSERT INTO opener_work (opener, requested_priority, requested_at," + " state) VALUES ('slate', 10, 1, 'complete')") + queue.close() + self.assertEqual(opener_completion_signal(sources), (1, 2)) + + def test_opener_completion_signal_is_none_when_the_queue_is_absent(self): + # None is "nothing is known", which the caller must read as "assume + # changed". A missing queue must not read as "no opener has finished". + sources = self._signal_sources( + os.path.join(self.temporary_directory.name, "gone.sqlite3")) + self.assertIsNone(opener_completion_signal(sources)) + + def test_opener_completion_signal_is_none_when_the_queue_is_unreadable(self): + # A file that is not a database reaches the query, not the connect, so + # this covers the second failure arm rather than the first. + directory = self.temporary_directory.name + queue_path = os.path.join(directory, "not_a_database.sqlite3") + with open(queue_path, "wb") as handle: + handle.write(b"this is not a SQLite file") + sources = self._signal_sources(queue_path) + self.assertIsNone(opener_completion_signal(sources)) + + def test_opener_completion_signal_survives_a_path_with_uri_syntax(self): + # A path holding '?' or '#' is a valid filename and a URI delimiter, so + # interpolating it truncates the path and opens a different database -- + # which returns None and silently disables revalidation. + directory = os.path.join(self.temporary_directory.name, "odd?dir#name") + os.makedirs(directory, exist_ok=True) + queue_path = os.path.join(directory, "queue.sqlite3") + queue = ERDQueue(queue_path) + queue._conn.execute( + "INSERT INTO opener_work (opener, requested_priority, requested_at," + " state) VALUES ('crane', 10, 1, 'complete')") + queue.close() + sources = self._signal_sources(queue_path) + self.assertEqual(opener_completion_signal(sources), (1, 1)) + def test_a_summary_of_branches_with_no_worker_status_stays_serializable(self): # branch_worker_status is NULL for a done or unqueued branch, and a # None key cannot be ordered against the string keys beside it. The diff --git a/tests/test_report_server.py b/tests/test_report_server.py index add37921..d781d302 100644 --- a/tests/test_report_server.py +++ b/tests/test_report_server.py @@ -653,6 +653,101 @@ def test_a_queue_backed_report_is_never_served_from_the_cache(self): self.assertEqual(self.calls, 3) + def test_a_waiter_does_not_file_the_builders_body_under_its_own_token(self): + # A waiter joins a build that started before its own signal read, so + # the body it receives can predate the token it holds. Filing under + # that token would publish a body older than the state the token names, + # and every later poll matching it would be served the stale ranking -- + # which inverts the guarantee that a stale signal costs freshness and + # never correctness. + started, release = Event(), Event() + bodies = iter(["first", "second"]) + + def collect_slowly(_sources, _request): + started.set() + release.wait(2) + self.calls += 1 + return {"data": next(bodies), "sources": {}} + + # The builder reads (1, 1); the waiter that joins mid-build reads (2, 2). + signals = iter([(1, 1), (2, 2), (2, 2), (2, 2)]) + with patch("report_server.collect_report", side_effect=collect_slowly), \ + patch("report_server.opener_completion_signal", + side_effect=lambda _sources: next(signals)): + with running_server(self.live_configuration) as base_url: + responses = [] + builder = Thread(target=lambda: responses.append( + request(base_url, "/api/view/leaderboard"))) + builder.start() + self.assertTrue(started.wait(1)) + waiter = Thread(target=lambda: responses.append( + request(base_url, "/api/view/leaderboard"))) + waiter.start() + time.sleep(0.05) + release.set() + builder.join(3) + waiter.join(3) + # A third request holding (2, 2) must not be handed the body + # built while the signal still read (1, 1). + _status, _headers, body = request( + base_url, "/api/view/leaderboard") + self.assertEqual(json.loads(body)["data"], "second") + + def test_a_build_that_recorded_a_source_error_is_not_cached(self): + # collect_leaderboard_report catches its own SQLite errors and returns + # an ordinary report with an empty ranking and the error on the source, + # so a degraded build is a normal-looking 200. Caching one pins an + # empty leaderboard until it expires, where every poll used to recover + # on the next request. + degraded = {"data": {"rows": []}, + "sources": {"cache": {"ok": False, "error": "disk I/O error"}}} + with patch("report_server.collect_report", + side_effect=lambda _s, _r: ( + setattr(self, "calls", self.calls + 1) or degraded)), \ + patch("report_server.opener_completion_signal", return_value=(3, 3)): + with running_server(self.live_configuration) as base_url: + self.get_leaderboard(base_url, times=3) + self.assertEqual(self.calls, 3) + + def test_a_source_that_was_never_consulted_does_not_block_caching(self): + # The leaderboard never opens the queue, so its queue source reports + # ok=false with no error. That is "not consulted", not a failure, and + # must not be read as one -- it would disable the cache entirely. + intact = {"data": {"rows": []}, + "sources": {"queue": {"ok": False, "error": None}, + "cache": {"ok": True, "error": None}}} + with patch("report_server.collect_report", + side_effect=lambda _s, _r: ( + setattr(self, "calls", self.calls + 1) or intact)), \ + patch("report_server.opener_completion_signal", return_value=(3, 3)): + with running_server(self.live_configuration) as base_url: + self.get_leaderboard(base_url, times=3) + self.assertEqual(self.calls, 1) + + def test_a_limit_evicted_by_newer_ones_is_rebuilt_rather_than_retained(self): + # The cache key is the request, and a request carries a user-controlled + # limit, so each distinct ?limit= value is a separate multi-megabyte + # body. Expiry alone only stops an entry being reused; without a cap + # the process grows until it is restarted. Eviction is observable as a + # rebuild: the earliest limit is gone once enough newer ones arrive. + cap = report_server.REPORT_CACHE_MAX_ENTRIES + with patch("report_server.collect_report", side_effect=self.collect_counting), \ + patch("report_server.opener_completion_signal", return_value=(3, 3)): + with running_server(self.live_configuration) as base_url: + for limit in range(1, cap + 1): + request(base_url, f"/api/view/leaderboard?limit={limit}") + self.assertEqual(self.calls, cap) + # Still held: re-asking for one of them rebuilds nothing. + request(base_url, "/api/view/leaderboard?limit=1") + self.assertEqual(self.calls, cap) + # Push past the cap, which evicts the oldest entries. + for limit in range(cap + 1, cap + 4): + request(base_url, f"/api/view/leaderboard?limit={limit}") + self.assertEqual(self.calls, cap + 3) + request(base_url, "/api/view/leaderboard?limit=2") + self.assertEqual(self.calls, cap + 4) + + class ReportServerMainTest(unittest.TestCase): def test_port_collision_exits_cleanly_instead_of_crashing(self): with running_server(fixture_configuration()) as base_url: From bf178f43a0aa7a6d1c0a180870c4876e13492b48 Mon Sep 17 00:00:00 2001 From: Sean Ahern Date: Sat, 19 Sep 2026 16:43:38 -0400 Subject: [PATCH 5/6] Bound the queue report's rows, and stop calling a busy worker dead Two report defects found while the reporter was serving live data. The queue report returned every branch it had registered when the caller named no limit, which the web client does not send by default. On the production queue that is 54,201 rows and a 68 MB payload: the request completes, so nothing errors, but the browser spends minutes fetching and rendering before it can show anything and the view reads as hung. Every other row-bearing report already defaults a limit; this one now defaults 100, and matched_rows still reports the full count so the view says how much it is not showing. A worker with a stale heartbeat was reported dead even while its process was running. The node counter that drives the heartbeat lives inside _BranchWorker._heartbeat, so a worker spending a long time inside one kernel call reports nothing for the whole of it -- observed live at 122.4M nodes and 0/s with the process at 74% CPU and a heartbeat 1,402 seconds old, against a 30-second liveness window. A fresh heartbeat still proves liveness on its own and costs no lookup; a stale one now consults the process table before the worker is called dead, matching the claims-not-heartbeats rule the scheduler already follows. The command line is checked with the pid, because pid numbers are reused and a recycled one would otherwise resurrect a departed worker. heartbeat_age_seconds is reported alongside, so "running but not reporting" is visible rather than inferred. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_01PpacbZRqT46b4Duhxq5QNQ --- report_model.py | 54 ++++++++++++++++++++++++++-- tests/test_report_model.py | 74 ++++++++++++++++++++++++++++++++++++++ 2 files changed, 125 insertions(+), 3 deletions(-) diff --git a/report_model.py b/report_model.py index a627846d..1ed63230 100644 --- a/report_model.py +++ b/report_model.py @@ -742,6 +742,38 @@ def _worker_number(worker_id): return worker_id.rsplit("-", 1)[-1] +#: Read from /proc to tell a live worker from a recycled pid. The swarm is +#: Linux-only, and the report runs on the same machine as the workers it +#: describes, so the process table is available and authoritative. +_WORKER_COMMAND_MARKER = "erd_search.py" + + +def _worker_process_is_running(pid): + """True when `pid` is a live swarm process on this machine. + + A heartbeat says when a worker last *reported*, which is not the same as + whether it is running. The node counter that drives the heartbeat lives + inside `_BranchWorker._heartbeat`, so a worker spending a long time inside + one kernel call reports nothing for the whole of it: observed live at + 122.4M nodes and 0/s with the process at 74% CPU and a heartbeat 1,402 + seconds old, against a 30-second liveness window. Calling that worker dead + asserts something the process table contradicts. + + The command line is checked as well as the pid, because pid numbers are + reused and a recycled number would otherwise resurrect a worker that is + genuinely gone. Anything unreadable answers False, which leaves the + heartbeat as the only evidence -- the behaviour from before this check + existed. + """ + if not pid: + return False + try: + with open(f"/proc/{int(pid)}/cmdline", "rb") as command_file: + return _WORKER_COMMAND_MARKER.encode() in command_file.read() + except (OSError, ValueError, TypeError): + return False + + def _normalize_worker(row, generated_at, answer_set): branch_key_value = _row_value(row, "current_branch_key") branch_key = bytes(branch_key_value) if branch_key_value is not None else None @@ -754,7 +786,13 @@ def _normalize_worker(row, generated_at, answer_set): "pid": row["pid"], "answer_count": _row_value(row, "n_words"), "updated_at": row["updated_at"], - "is_live": generated_at - row["updated_at"] <= WORKER_LIVENESS_SECONDS, + # A fresh heartbeat proves liveness; a stale one does not disprove it, + # so the process table is consulted before a worker is called dead. + "is_live": ( + generated_at - row["updated_at"] <= WORKER_LIVENESS_SECONDS + or _worker_process_is_running(row["pid"]) + ), + "heartbeat_age_seconds": max(0, generated_at - row["updated_at"]), "branch_reference": branch_reference(branch_key) if branch_key else None, "branch_key_hex": branch_key.hex() if branch_key else None, "branch_context": _normalized_branch_spine(row, answer_set), @@ -2868,6 +2906,16 @@ def _row_matches_branch_target(row, branch_target, prefix): NO_WORKER_STATUS_BUCKET = "not_applicable" +#: Rows the queue report returns when the caller names no limit. Every other +#: row-bearing report already defaults one; without it the queue returns every +#: branch it has ever registered -- 54,201 rows and a 68 MB payload on the +#: production queue, which the browser spends minutes fetching and rendering +#: before showing anything. `matched_rows` still reports the full count, so +#: the view says how much it is not showing, and the row-limit control raises +#: it. +DEFAULT_QUEUE_ROW_LIMIT = 100 + + def _collection_summary(rows): by_status = {} by_worker_status = {} @@ -2910,8 +2958,8 @@ def collect_queue_report(sources: ReportOpeners, request: ReportRequest) -> dict rows, _prefix = _scoped_queue_rows(queue, request) data["summary"] = _collection_summary(rows) data["matched_rows"] = len(rows) - limit = request.filters.limit - data["rows"] = rows[:limit] if limit is not None else rows + limit = request.filters.limit or DEFAULT_QUEUE_ROW_LIMIT + data["rows"] = rows[:limit] for row in data["rows"]: row["branch_reference"] = branch_reference(bytes(row.pop("branch_key"))) row["spine"] = _normalized_branch_spine(row, answer_set) diff --git a/tests/test_report_model.py b/tests/test_report_model.py index 86b3725d..2dce1a3c 100644 --- a/tests/test_report_model.py +++ b/tests/test_report_model.py @@ -15,6 +15,9 @@ import erd_search import report_model from report_model import ( + _worker_process_is_running, + _normalize_worker, + DEFAULT_QUEUE_ROW_LIMIT, opener_completion_signal, _collection_summary, NO_WORKER_STATUS_BUCKET, @@ -2040,6 +2043,77 @@ def test_opener_completion_signal_survives_a_path_with_uri_syntax(self): sources = self._signal_sources(queue_path) self.assertEqual(opener_completion_signal(sources), (1, 1)) + def test_a_stale_heartbeat_does_not_make_a_running_worker_dead(self): + # The node counter that drives the heartbeat lives inside _heartbeat, + # so a worker spending a long time in one kernel call reports nothing + # for the whole of it. Observed live at 122.4M nodes and 0/s with the + # process at 74% CPU and a heartbeat 1,402s old against a 30s window. + row = {"worker_id": "worker-0", "pid": 4242, "updated_at": 1_000, + "current_branch_key": None, "cur_candidate": None, + "best_guess": None, "n_words": None, "claim_idx": None, + "claim_started_at": None, "claims_done": 0, "cur_max_depth": None, + "cur_nodes": None, "cur_help_depth": 0, "node_rate": None} + generated_at = 1_000 + WORKER_LIVENESS_SECONDS + 1_372 + with patch("report_model._worker_process_is_running", return_value=True): + worker = _normalize_worker(row, generated_at, set()) + self.assertTrue(worker["is_live"]) + self.assertEqual(worker["heartbeat_age_seconds"], + WORKER_LIVENESS_SECONDS + 1_372) + + with patch("report_model._worker_process_is_running", return_value=False): + gone = _normalize_worker(row, generated_at, set()) + self.assertFalse(gone["is_live"]) + + def test_a_fresh_heartbeat_needs_no_process_lookup(self): + # The heartbeat is the cheap evidence and it is sufficient on its own, + # so the common case must not pay for a /proc read per worker. + row = {"worker_id": "worker-1", "pid": 4242, "updated_at": 1_000, + "current_branch_key": None, "cur_candidate": None, + "best_guess": None, "n_words": None, "claim_idx": None, + "claim_started_at": None, "claims_done": 0, "cur_max_depth": None, + "cur_nodes": None, "cur_help_depth": 0, "node_rate": None} + with patch("report_model._worker_process_is_running") as looked_up: + worker = _normalize_worker(row, 1_001, set()) + self.assertTrue(worker["is_live"]) + looked_up.assert_not_called() + + def test_a_recycled_pid_does_not_resurrect_a_departed_worker(self): + # This process is alive and is not a swarm worker, so a pid that has + # been reused must not read as the worker that once held it. + self.assertFalse(_worker_process_is_running(os.getpid())) + self.assertFalse(_worker_process_is_running(None)) + # A pid that cannot exist, and one that is not a number at all. + self.assertFalse(_worker_process_is_running(2 ** 31 - 1)) + self.assertFalse(_worker_process_is_running("not-a-pid")) + + def test_the_queue_report_limits_its_rows_when_the_caller_names_no_limit(self): + # Without a default the queue returns every branch it has registered. + # On the production queue that is 54,201 rows and 68 MB, which the + # browser spends minutes fetching before it can draw anything -- + # the report never fails, it just never arrives. + queue = self._open_queue() + queue.add_pending_many([ + (ScoreCache.encode_subset(["crane", f"w{index:04d}x"]), 2, 9, "salet", 0) + for index in range(DEFAULT_QUEUE_ROW_LIMIT + 5) + ]) + queue.close() + data = collect_report(self.sources, ReportRequest( + report_kind="queue"))["data"] + self.assertEqual(len(data["rows"]), DEFAULT_QUEUE_ROW_LIMIT) + self.assertGreater(data["matched_rows"], DEFAULT_QUEUE_ROW_LIMIT) + + def test_an_explicit_queue_limit_still_wins(self): + queue = self._open_queue() + queue.add_pending_many([ + (ScoreCache.encode_subset(["crane", f"w{index:04d}x"]), 2, 9, "salet", 0) + for index in range(20) + ]) + queue.close() + data = collect_report(self.sources, ReportRequest( + report_kind="queue", filters=ReportFilters(limit=7)))["data"] + self.assertEqual(len(data["rows"]), 7) + self.assertEqual(data["matched_rows"], 20) + def test_a_summary_of_branches_with_no_worker_status_stays_serializable(self): # branch_worker_status is NULL for a done or unqueued branch, and a # None key cannot be ordered against the string keys beside it. The From 83babdf97d65b13d4bcc0196e542d3270b17704c Mon Sep 17 00:00:00 2001 From: Sean Ahern Date: Sat, 19 Sep 2026 23:46:28 -0400 Subject: [PATCH 6/6] Publish a leaderboard body before releasing the waiters on it The builder released waiters and dropped its in-flight marker before storing the entry, leaving a window in which the build was finished, the marker was gone and the cache was still empty. A request arriving there found neither and started a second build of a report already in hand. Wrong only in cost, never in content -- but the fix removes a branch rather than adding one, so it is cheaper than the race it closes. The window is a dict write while reaching it takes an HTTP round trip, so it is not reachable by timing alone: a first attempt at a guard passed against the old ordering in 12 of 12 runs and proved nothing. The test now widens exactly the window in question, by slowing release of the lock the marker is dropped under, which is the last thing the old ordering did before publishing. It catches the old ordering in 10 of 10 runs and passes the new one in 10 of 10. Under the correct ordering the entry is already stored by then, so widening changes nothing -- which is the property being asserted. Reported by Codex on PR #364. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_01PpacbZRqT46b4Duhxq5QNQ --- report_server.py | 11 +++++-- tests/test_report_server.py | 59 +++++++++++++++++++++++++++++++++++++ 2 files changed, 68 insertions(+), 2 deletions(-) diff --git a/report_server.py b/report_server.py index a3e0aa03..f1c72f42 100644 --- a/report_server.py +++ b/report_server.py @@ -518,6 +518,15 @@ def collect_report_once(request, token, started_at): report = collect_report(configuration.sources, request) in_flight.body = encode_report(report) in_flight.intact = report_is_intact(report) + # Published before the waiters are released and before the + # in-flight marker is dropped, so no arriving request can find + # neither. Releasing first leaves a window in which the build + # is finished, the marker is gone and the entry is not yet + # stored, and a request landing there starts a second build of + # a report that is already in hand. + if token is not None and in_flight.intact: + store_cached_body( + request, token, started_at, in_flight.body) except Exception as error: in_flight.error = error finally: @@ -528,8 +537,6 @@ def collect_report_once(request, token, started_at): in_flight.completed.wait() if in_flight.error is not None: raise in_flight.error - if is_builder and token is not None and in_flight.intact: - store_cached_body(request, token, started_at, in_flight.body) return in_flight.body class ReportHandler(BaseHTTPRequestHandler): diff --git a/tests/test_report_server.py b/tests/test_report_server.py index d781d302..141abfeb 100644 --- a/tests/test_report_server.py +++ b/tests/test_report_server.py @@ -6,6 +6,7 @@ from threading import Event, Lock, Thread import io import json +import threading import os import tempfile import time @@ -693,6 +694,64 @@ def collect_slowly(_sources, _request): base_url, "/api/view/leaderboard") self.assertEqual(json.loads(body)["data"], "second") + def test_a_waiter_is_released_only_once_the_body_is_published(self): + # Releasing waiters before publishing leaves a window in which the + # build is finished, the in-flight marker is gone, and the entry is not + # yet stored. A request landing there finds neither and rebuilds a + # report already in hand. + # + # That window is a dict write, and reaching it takes an HTTP round + # trip, so it is not reachable by timing alone -- this test widens it + # by slowing the release of the lock the marker is dropped under, which + # is the last thing the old ordering did before publishing. Under the + # correct ordering the entry is already stored by then, so widening + # changes nothing. + started, release = Event(), Event() + widen = [False] + + class SlowReleaseLock: + def __init__(self): + self._lock = threading.Lock() + + def __enter__(self): + return self._lock.__enter__() + + def __exit__(self, *details): + result = self._lock.__exit__(*details) + if widen[0]: + time.sleep(0.4) + return result + + def collect_slowly(_sources, _request): + self.calls += 1 + started.set() + release.wait(2) + return self.report + + followup = [] + with patch("report_server.Lock", SlowReleaseLock), \ + patch("report_server.collect_report", side_effect=collect_slowly), \ + patch("report_server.opener_completion_signal", return_value=(5, 5)): + with running_server(self.live_configuration) as base_url: + builder = Thread(target=lambda: request( + base_url, "/api/view/leaderboard")) + builder.start() + self.assertTrue(started.wait(1)) + + def wait_then_ask(): + request(base_url, "/api/view/leaderboard") + followup.append(request(base_url, "/api/view/leaderboard")) + + waiter = Thread(target=wait_then_ask) + waiter.start() + time.sleep(0.05) + widen[0] = True + release.set() + builder.join(5) + waiter.join(5) + self.assertEqual([status for status, _h, _b in followup], [200]) + self.assertEqual(self.calls, 1) + def test_a_build_that_recorded_a_source_error_is_not_cached(self): # collect_leaderboard_report catches its own SQLite errors and returns # an ordinary report with an empty ranking and the error on the source,