From e9a1677918a8f17066d234d397066bbb579b0417 Mon Sep 17 00:00:00 2001 From: Jules GRIVOT PELISSON Date: Thu, 29 Oct 2026 22:29:18 -0400 Subject: [PATCH 1/2] FIX: list runs incrementally off the event loop, refuse uploaded model weights and cap each workstation's runs in flight --- RUN_PROGRESS.md | 14 + docker-compose.yml | 13 + server/.env.example | 11 + server/benchmark_client.py | 15 +- server/benchmark_jobs.py | 25 ++ server/config.py | 15 ++ server/deployment.toml | 8 +- server/main.py | 328 +++++++++++++++++++----- server/registry/conventions.py | 20 ++ server/telemetry.py | 32 ++- server/tests/test_compose_logging.py | 35 +++ server/tests/test_dashboard_data.py | 54 ++++ server/tests/test_detached.py | 34 +++ server/tests/test_file_arguments.py | 97 +++++++ server/tests/test_main.py | 84 ++++++ server/tests/test_page_polling.py | 137 ++++++++++ server/tests/test_runs.py | 277 ++++++++++++++++++++ server/tests/test_runs_per_client.py | 295 +++++++++++++++++++++ server/tests/test_telemetry.py | 40 +++ server/tests/test_upload_part_limits.py | 35 +++ server/wire/debug_page.py | 16 +- server/wire/runs.py | 156 ++++++++++- server/wire/status_page.py | 18 +- server/wire/transfer.py | 8 +- 24 files changed, 1678 insertions(+), 89 deletions(-) create mode 100644 server/tests/test_compose_logging.py create mode 100644 server/tests/test_page_polling.py create mode 100644 server/tests/test_runs_per_client.py diff --git a/RUN_PROGRESS.md b/RUN_PROGRESS.md index f7fa2c9..190d57e 100644 --- a/RUN_PROGRESS.md +++ b/RUN_PROGRESS.md @@ -105,6 +105,13 @@ delivered, or when the run directory disappears. events with `"kind": "log"` (ยง4e). **Without it there are none**: a client released before log lines existed would draw one as progress. +After 15 seconds with nothing to send, the stream sends an SSE comment, +`: ping`. A tool can be silent for minutes -- a registration, a model loading -- +and a client's read timeout is shorter than that; without the comment it +dropped the stream and reconnected, replaying every event from the start. A +line starting with `:` is never an event, to the Slicer client or to any SSE +parser. + The last header is not decoration: nginx buffers a proxied response by default, which for a stream means the client sees nothing until the run ends - the exact failure the endpoint exists to prevent. `curl` needs `-N` for the same reason. @@ -369,6 +376,13 @@ Every append and every read stamps the directory, so `RUN_TTL_SECONDS` is an progress for six hours is never reaped under itself. The same `_reaper_loop` that sweeps transfers sweeps these. +A run a previous server process left `pending` or `running` -- `TEMP_DIR` +outlives a restart -- is ended `failed` when the next one starts, with the +message "The server restarted during this run.": nobody else would ever end +it, and its client waits on exactly that terminal event. Its idle clock is +kept as it was. A `paused` run is left alone, since a resume picks it up +whichever process serves it. This assumes the single process `serve.py` runs. + The reaper is a safety net rather than the normal path. **A run is removed with the request that owned it**, on success (as a background task, so the directory survives until the response has finished streaming and a watcher has the whole diff --git a/docker-compose.yml b/docker-compose.yml index 7e89f02..1e7a0a6 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -5,6 +5,17 @@ # Pinned, the stack keeps one identity wherever it is cloned. name: visor-serve +# The log every server service keeps: five files of 20 MB, the oldest dropped. +# Docker's default json-file driver keeps everything a container ever wrote for +# as long as it exists, and a server up for months -- an access line per panel +# poll, every two seconds -- fills the disk TEMP_DIR lives on, where a run then +# fails to stage its upload for a reason nobody looks for in a log. +x-bounded-log: &bounded-log + driver: json-file + options: + max-size: "20m" + max-file: "5" + # Shared by the two server services below, which differ only in whether they # get the GPU. Two services rather than one, because a compose device # reservation is ALL-OR-NOTHING: a service asking for an nvidia device cannot @@ -20,6 +31,7 @@ x-inference-base: &inference-base image: ${VISOR_SERVER_IMAGE:-ghcr.io/dcbia-ortholab/visor-serve:main} shm_size: 8gb working_dir: /workspace/server + logging: *bounded-log volumes: # See `inference-venvs`: a public certificate, when TLS_CERT_FILE names one. - ./tls:/etc/visor-tls:ro @@ -282,6 +294,7 @@ services: # block as `inference`. inference-venvs: profiles: ["venvs"] + logging: *bounded-log build: context: . dockerfile: docker/Dockerfile diff --git a/server/.env.example b/server/.env.example index 5a6eab0..098d143 100644 --- a/server/.env.example +++ b/server/.env.example @@ -324,3 +324,14 @@ RUN_CANCEL_POLL_SECONDS=1.0 # and the rest are dropped, terminal events always excepted so a watcher's # stream can still end. MAX_RUN_EVENTS=2000 + +# --- one workstation's share of the queue ----------------------------------- + +# Runs one workstation -- one client address, as the server sees it -- may have +# in flight at once: registered and not yet ended, waiting for the card +# included. A run stopped at a checkpoint does not count; it waits for a +# person. Past it POST /run answers 429 before reading a byte. Every +# workstation holds the same API token, so without it one client could fill +# the queue the others wait in. 0 means no cap. Behind a proxy that is not in +# FORWARDED_ALLOW_IPS every workstation IS the proxy, and they share one cap. +MAX_RUNS_PER_CLIENT=10 diff --git a/server/benchmark_client.py b/server/benchmark_client.py index 2b4c837..0c536cd 100644 --- a/server/benchmark_client.py +++ b/server/benchmark_client.py @@ -55,6 +55,11 @@ from typing import Optional RUN_ID_HEADER = "X-Run-Id" +# This battery's key, which the server minted for it and checks on every +# POST /run so a battery is not refused as one workstation sending too many +# runs (benchmark_jobs.KEY_HEADER, which this script cannot import). +BENCHMARK_KEY_HEADER = "X-Benchmark-Key" +BENCHMARK_KEY_ENV = "SADT_BENCHMARK_KEY" # Long: a cohort segmentation legitimately takes an hour, and a benchmark that # times out before the tool does would record a failure that did not happen. REQUEST_TIMEOUT = 3 * 3600 @@ -138,6 +143,14 @@ def _request(url: str, token: str, data=None, method=None, timeout=POLL_TIMEOUT, return response.status, response.read(), dict(response.headers) +def _run_headers(run_id: str) -> dict: + headers = {RUN_ID_HEADER: run_id} + key = os.environ.get(BENCHMARK_KEY_ENV) + if key: + headers[BENCHMARK_KEY_HEADER] = key + return headers + + def _send(url: str, token: str, body: bytes, method: str, content_type: str, headers=None): request = urllib.request.Request(url, data=body, method=method) @@ -332,7 +345,7 @@ def _one_run(base: str, token: str, item: dict, client: int, origin: float, status, body, headers = _request( f"{base}/run/{urllib.parse.quote(item['tool'])}", token, data=form, method="POST", - timeout=REQUEST_TIMEOUT, headers={RUN_ID_HEADER: run_id}) + timeout=REQUEST_TIMEOUT, headers=_run_headers(run_id)) _record(LEDGER_END, run_id) record["seconds"] = round(time.time() - posted, 2) record["invocation"]["produced"] = {"bytes": len(body)} diff --git a/server/benchmark_jobs.py b/server/benchmark_jobs.py index 9ecad7a..d482328 100644 --- a/server/benchmark_jobs.py +++ b/server/benchmark_jobs.py @@ -35,9 +35,11 @@ from __future__ import annotations +import hmac import json import logging import os +import secrets import shutil import signal import subprocess @@ -63,6 +65,15 @@ LATE_SWEEP_SECONDS = 10.0 _LATE_SWEEP_POLL = 0.25 +# How a battery's runs say they are the battery's: a key minted per battery, +# handed to its process alone and sent beside each POST /run, so the cap on +# runs per workstation (settings.MAX_RUNS_PER_CLIENT) does not refuse the very +# concurrency a battery exists to measure. Never in this process's own +# environment, so no tool inherits it, and never in `current()`. Repeated in +# benchmark_client.py, which runs as a script and imports nothing from here. +KEY_HEADER = "X-Benchmark-Key" +KEY_ENV = "SADT_BENCHMARK_KEY" + _lock = threading.Lock() _current: Optional[dict] = None # Serialises settling a battery's runs: a stop and the watcher both settle, @@ -102,6 +113,17 @@ def _finished(job: dict) -> bool: return process is None or process.poll() is not None +def is_battery(presented) -> bool: + """Whether `presented` is the key of the battery running now.""" + if not presented: + return False + with _lock: + job = _current + if job is None or not job.get("_key") or _finished(job): + return False + return hmac.compare_digest(str(presented), job["_key"]) + + def start(plan: dict, summary_dir: str, base_url: str, token: str, scratch_dir: str) -> dict: """Spawn the battery. Raises `BatteryError` if one is already running.""" @@ -131,8 +153,10 @@ def start(plan: dict, summary_dir: str, base_url: str, token: str, with open(plan_path, "w", encoding="utf-8") as handle: json.dump(plan, handle) + key = secrets.token_urlsafe(32) environment = dict(os.environ) environment["API_TOKEN"] = token + environment[KEY_ENV] = key environment["SADT_TLS_CAFILE"] = tls.certificate_paths()[0] client = os.path.join(os.path.dirname(os.path.abspath(__file__)), "benchmark_client.py") @@ -155,6 +179,7 @@ def start(plan: dict, summary_dir: str, base_url: str, token: str, "exit_code": None, "cancelled_runs": None, "_process": process, + "_key": key, "_plan_path": plan_path, "_runs_path": runs_path, } diff --git a/server/config.py b/server/config.py index f2784c3..a89a119 100644 --- a/server/config.py +++ b/server/config.py @@ -341,6 +341,21 @@ def tool_roots(self) -> tuple: # fill TEMP_DIR; past the cap one "not reported" event is delivered and the # rest are dropped, terminal events always excepted. MAX_RUN_EVENTS: int = 2000 + # Runs one workstation may have in flight at once -- registered and not yet + # ended, waiting for the card included -- past which POST /run answers 429 + # before it reads a byte. Every workstation holds the same API token, so + # without this one client sending sixty runs filled the queue every other + # one waits in. A PAUSED run does not count: it waits for a person and + # holds no machine. 0 means no cap. + # + # 10 is well above what the Slicer client sends: one run per panel at a + # time (CONCURRENT_RUNS = 1), and on a workstation an operator set to + # parallel batches as many as the budget runs side by side (`max_parallel` + # of GET /clients/me, CPUs / CPUs per job). Counted per address as the + # server sees it: behind a proxy that is not in FORWARDED_ALLOW_IPS every + # workstation IS the proxy, and they share one cap. A benchmark battery an + # operator starts is not counted. + MAX_RUNS_PER_CLIENT: int = 10 @field_validator("SADT_DISPATCH_MODE") diff --git a/server/deployment.toml b/server/deployment.toml index 50566a1..e3ad14a 100644 --- a/server/deployment.toml +++ b/server/deployment.toml @@ -204,8 +204,8 @@ width_from = "structures" # AMASSS has one model and resolves it itself when `model` is empty, from # DATA/AMASSS/models/AMASSS_Models. The convention would fill the argument with # the models FOLDER, which is not a bundle, and would put a picker with a -# single answer in front of a clinician. So it is neither filled nor asked; -# a caller that names a bundle is still obeyed. +# single answer in front of a clinician. So it is neither filled nor asked -- +# nor uploaded: a model never is (main.py `_reject_upload_for_model`). server_selectable = { model = "none" } hidden = ["model"] @@ -229,8 +229,8 @@ dispatch = { "CBCT" = "AREG_CBCT", "IOS" = "AREG_IOS", "IOS to CBCT" = "AREG_IOS # convention that reads `*_model` as "pick one from DATA/ASO/models/" would fill # it with ASO's own model folder and pass that on as ALI's bundle. Measured in a # real run on 2026-09-11: ALI received `DATA/ASO/models` and predicted with the -# wrong weights, silently. "none" opts it out; a caller may still name a bundle -# to pin which weights ran. +# wrong weights, silently. "none" opts it out, and the panel no longer asks for +# it: a model is never uploaded (main.py `_reject_upload_for_model`). server_selectable = { landmark_model = "none" } [tools.ALI] diff --git a/server/main.py b/server/main.py index d294a34..d6df8cb 100644 --- a/server/main.py +++ b/server/main.py @@ -18,7 +18,9 @@ import secrets import shutil import subprocess +import sys import tempfile +import threading import time import traceback import zlib @@ -42,6 +44,7 @@ # server -- but the other direction is safe and is what lets the rewind # edit the same record the runner wrote. It is standard library only. from execution import admission, costs, dispatch, reports, runner +from execution.dispatch import ToolExecutionError from registry import conventions, facade from registry.facade import FacadeTool import file_utils @@ -95,16 +98,38 @@ async def _reaper_loop() -> None: logger.exception("reaper sweep failed") +# The card's last reading for the pages and the trace: (monotonic time, (total, +# free)), or None before the first. Each reading is an `nvidia-smi` exec, and +# `/status` and the panel each took one per poll -- a process every two seconds +# per open page, on the machine whose card the runs are waiting for. A gauge +# drawn from a reading a few seconds old is the same gauge. ADMISSION does not +# read through this: deciding whether a job fits wants the card as it is now. +_card_reading: Optional[tuple] = None +_card_lock = threading.Lock() + + +def _card() -> tuple: + """`(total, free)` VRAM, read at most once every `telemetry.TRACE_SECONDS`; + `(None, None)` when there is no card to read.""" + global _card_reading + with _card_lock: + now = time.monotonic() + if _card_reading is not None and now - _card_reading[0] < telemetry.TRACE_SECONDS: + return _card_reading[1] + try: + card = resources.detect_vram_bytes() + except Exception: # noqa: BLE001 - no card is a missing line, not an error + card = (None, None) + _card_reading = (now, card) + return card + + async def _trace_loop() -> None: """Sample the machine for the operator page's graphs, for as long as the server runs. One point every `telemetry.TRACE_SECONDS`, off the event loop because reading the card is a subprocess.""" def take() -> None: - try: - card = resources.detect_vram_bytes() - except Exception: # noqa: BLE001 - no card is a missing line, not an error - card = (None, None) - telemetry.sample_trace(admission.budget().snapshot(), card) + telemetry.sample_trace(admission.budget().snapshot(), _card()) while True: try: @@ -137,6 +162,10 @@ async def _lifespan(_app: FastAPI): if restored: logger.info("Run history: %d finished run(s) read back from %s", restored, settings.HISTORY_DIR) + # Before any request is served: a run the previous process left pending or + # running has nobody left to end it, and its client is waiting for exactly + # that (see runs.fail_orphans, which assumes serve.py's single process). + await anyio.to_thread.run_sync(runs.fail_orphans) async with anyio.create_task_group() as task_group: task_group.start_soon(_reaper_loop) task_group.start_soon(_trace_loop) @@ -479,11 +508,43 @@ def _temp_root_of(path: str) -> Optional[str]: def _discard(work_dir: Optional[str], scratch_dirs: list) -> None: """Remove everything this request created, right now. Used on the error - paths, where no response will ever stream and background tasks won't run.""" + paths, where no response will ever stream and background tasks won't run. + Called through `_off_loop`: it is a tree of confidential imaging.""" for directory in ([work_dir] if work_dir else []) + list(scratch_dirs): shutil.rmtree(directory, ignore_errors=True) +def _remove_staged(input_paths, resolved_files) -> None: + """Delete a request's staged inputs once the tool is done with them. + + Uploaded inputs go; server-side data (DATA_DIR) is persistent and must + never be deleted, so of what the data store resolved only a backend's + temporary copy does (see ResolvedFile.is_temporary in data_store.py).""" + for input_path in input_paths: + if os.path.exists(input_path): + os.remove(input_path) + for resolved in resolved_files: + if resolved.is_temporary and os.path.exists(resolved.path): + os.remove(resolved.path) + + +async def _off_loop(function, *args): + """`function(*args)` in a worker thread, finished even if the request is + cancelled. + + For the disk work a request does on its way out -- deleting what it + staged, writing a run's terminal event. Deleting a staged CBCT costs about + 105 ms a gigabyte, and reading back a chatty tool's whole event file half + a second; on the event loop either froze every other request for as long. + Shielded because these run in `except` and `finally` blocks, very often + BECAUSE the request was cancelled (a client that hung up), and a cleanup + that a cancellation could interrupt would leave a patient's scan on disk + or a run that never ended. Keyword arguments go in a functools.partial. + """ + with anyio.CancelScope(shield=True): + return await anyio.to_thread.run_sync(function, *args) + + def _media_type_of(path: str) -> str: """Content-Type for a file about to be streamed. Derived from the real extension so an .xlsx is never mislabeled as a generic zip; the @@ -816,13 +877,21 @@ def server_status() -> dict: name the file it is working on, which is a patient's. Phases and counts say what the server is doing without saying whose data it is doing it to. """ + return _status_report(runs.active()) + + +def _status_report(listing: list) -> dict: + """`/status`, built around a run listing the caller already took. + + Split out so the admin panel, which wants the listing WITH each run's + chain, lists the runs once rather than twice: it used to call + `server_status()` and then replace the masked listing it carried, so every + poll paid for a listing nobody read. + """ allocation = resources.allocation() budget = admission.budget() - free_vram, total_vram = None, None - try: - total_vram, free_vram = resources.detect_vram_bytes() - except Exception: # noqa: BLE001 - a status page must not fail on a probe - pass + # A status page must not fail on a probe, and `_card` never raises. + total_vram, free_vram = _card() learned = costs.known() return _masked_run_ids({ "budget": { @@ -843,7 +912,7 @@ def server_status() -> dict: # without the server ever restarting. "maintenance": maintenance.snapshot(), "card": {"free_bytes": free_vram, "total_bytes": total_vram}, - "runs": runs.active(), + "runs": listing, "costs": { name: { "vram_bytes": cost.vram_bytes, @@ -1120,7 +1189,7 @@ def server_debug_data() -> dict: """Everything `/status` reports, plus what the machine says about itself. A superset rather than a second opinion: the budget, the admission holds - and the learned costs are read through `server_status` so the two endpoints + and the learned costs are read through `_status_report` so the two endpoints cannot drift into disagreeing about the same number. What is added here is the half that `/status` deliberately does not carry -- live CPU, live host memory, disk, in-flight requests, and the queue log -- because it is @@ -1131,7 +1200,12 @@ def server_debug_data() -> dict: patient's. Phases and counts say what the machine is doing without saying whose data it is doing it to. """ - report = server_status() + # The live listing with each run's open nested calls: `/status` keeps the + # narrower one it always had, and only this page needs to say that an + # AREG is, right now, inside ASO inside ALI_CBCT. Taken once, and the + # status report is built around it, so a poll lists the runs once. + listing = runs.active(with_chain=True) + report = _status_report(listing) allocation = resources.allocation() try: node = os.uname().nodename @@ -1162,10 +1236,9 @@ def server_debug_data() -> dict: # offer them or say why it does not. Never the token itself. "admin_enabled": bool(settings.ADMIN_TOKEN), } - # The live listing again, with each run's open nested calls: `/status` - # keeps the narrower one it always had, and only this page needs to say - # that an AREG is, right now, inside ASO inside ALI_CBCT. - report["runs"] = runs.active(with_chain=True) + # Unmasked: the panel is behind the admin token, and its controls act on + # the real ids. The status report above carried them masked. + report["runs"] = listing report["trace"] = telemetry.trace(since=time.time() - TRACE_WINDOW_SECONDS) report["cpu_percent"] = telemetry.cpu_percent() report["ram"] = telemetry.ram() @@ -1176,9 +1249,13 @@ def server_debug_data() -> dict: # across every worker, while `ledger` is this process's memory of what it # admitted and what that cost. The page joins them on `run_id`, and a run # present in one and not the other is a fact worth being able to see. - report["ledger"] = telemetry.run_ledger() + # + # Copied out once: the page's rows are the newest of the same history the + # client view counts over, which `_client_activity` only reads. + history = telemetry.run_ledger(limit=telemetry.LEDGER_SIZE) + report["ledger"] = history[:telemetry.LEDGER_PAGE] report["uptime"] = telemetry.tool_uptime() - report["clients"] = _client_activity(report["runs"], telemetry.run_ledger(limit=telemetry.LEDGER_SIZE)) + report["clients"] = _client_activity(report["runs"], history) # TEMP_DIR is where uploads, results and run directories live -- the space a # download actually costs this machine. DATA_DIR is mounted read-only and # cannot grow, but it is the other half of "what is this disk holding". @@ -2163,14 +2240,20 @@ async def upload_status(upload_id: str) -> dict: } -async def _read_bounded(request: Request, limit: int) -> bytes: +async def _read_bounded(request: Request, limit: int) -> bytearray: + """The body, refused past `limit` bytes, as the bytearray it was read into. + + Not converted to `bytes`: that is a second full copy of a part of up to + 64 MB, per part, with a client sending several at once. Everything it is + handed to -- the gunzip, the checksum, `pwrite` -- reads a bytearray as is. + """ received = bytearray() async for piece in request.stream(): received += piece if len(received) > limit: raise HTTPException(status_code=status.HTTP_413_CONTENT_TOO_LARGE, detail=f"A part is at most {limit} bytes.") - return bytes(received) + return received def _gunzip_bounded(data: bytes, limit: int) -> bytes: @@ -2331,6 +2414,14 @@ async def run_snapshot(run_id: str, logs: str = "") -> dict: raise _run_error(exc) +# How long the event stream may say nothing before it says so. A tool can be +# silent for minutes -- a registration, a model loading -- and the client's +# read timeout is 30 s: past it the client dropped the stream, reconnected, +# and replayed every event from the start. Half that, so one lost comment +# still leaves the next in time. +_SSE_HEARTBEAT_SECONDS = 15 + + @app.get("/runs/{run_id}/events", dependencies=[Depends(verify_token)]) async def run_events(run_id: str, logs: str = "") -> StreamingResponse: """Server-Sent Events, oldest first, INCLUDING what was written before this @@ -2358,8 +2449,10 @@ async def run_events(run_id: str, logs: str = "") -> StreamingResponse: async def frames(): reader = runs.EventReader(directory, logs=audiences) + quiet = 0.0 while True: - for event in await anyio.to_thread.run_sync(reader.read): + events = await anyio.to_thread.run_sync(reader.read) + for event in events: yield f"data: {json.dumps(event)}\n\n".encode("utf-8") if reader.finished: # A terminal event was delivered, or the directory went away @@ -2367,7 +2460,15 @@ async def frames(): # and holding the connection open would only look like one # still going. return + if events: + quiet = 0.0 + elif quiet >= _SSE_HEARTBEAT_SECONDS: + # An SSE comment: a line starting with ':' is never an event, + # to the client's parser or to any other. + yield b": ping\n\n" + quiet = 0.0 await anyio.sleep(settings.RUN_EVENT_POLL_SECONDS) + quiet += settings.RUN_EVENT_POLL_SECONDS return StreamingResponse( frames(), @@ -2611,6 +2712,24 @@ def _reject_upload_for_scalar(spec, field_name: str) -> None: ) +def _reject_upload_for_model(field_name: str) -> None: + """A model is never uploaded, whatever the deployment opted it out of. + + `server_selectable = { _model = "none" }` says the tool finds its own + weights, and leaves the argument a plain `path` -- which made it an upload. + A model loader unpickles what it is handed, so an uploaded model is code + run on this server (AREG_IOS loaded its checkpoint with `torch.load`). + Refused before a byte is staged. `*_reference` is not a model: see + `conventions.WEIGHTS_NAME`. + """ + if conventions.names_weights(field_name): + raise HTTPException( + status_code=status.HTTP_422_UNPROCESSABLE_CONTENT, + detail=(f"'{field_name}' is a model: the server uses its own weights, " + "they cannot be uploaded."), + ) + + def _unpacks_to_a_folder(kind: str, extension: str) -> bool: """Does this input arrive as an archive that must be extracted first? @@ -2669,6 +2788,94 @@ def _registered_run(request: Request, tool_name: str) -> Optional[str]: raise _run_error(exc) +# Runs this process is serving that no registry entry stands for -- a request +# that sent no `X-Run-Id` -- per client address. Taken by `_claimed_run`, given +# back by `_unwatched_run` in a `finally`, so no ending can leak a slot; and in +# memory, so a restart cannot leave a stale count behind it. +_unwatched_runs: dict = {} +_unwatched_lock = threading.Lock() + +# Held across the count and the registration (or the unwatched slot), so two +# requests cannot both count the same free slot and both go past the cap. A +# threading lock, not an anyio one: the step runs in a worker thread, and a +# module-level anyio lock binds to the first event loop that uses it. +_claim_lock = threading.Lock() + + +def _claimed_run(request: Request, tool_name: str, address: Optional[str], + detached: bool) -> Optional[str]: + """Refuse past the cap, then register the run -- one step, under + `_claim_lock`, in a worker thread: counting lists every run on disk, and + on the event loop that held every other request for as long. + + Returns the run id, or None for a run sent without one; such a run then + holds an unwatched slot, which `_unwatched_run` gives back. A detached run + without an id takes none: it is refused for lacking the id. + """ + with _claim_lock: + _refuse_past_the_client_cap(request, address) + run_id = _registered_run(request, tool_name) + if run_id is None and not detached: + with _unwatched_lock: + _unwatched_runs[address] = _unwatched_runs.get(address, 0) + 1 + return run_id + + +@contextlib.contextmanager +def _unwatched_run(address: Optional[str]): + """Give back the slot `_claimed_run` took, however the run ends.""" + try: + yield + finally: + with _unwatched_lock: + left = _unwatched_runs.get(address, 1) - 1 + if left > 0: + _unwatched_runs[address] = left + else: + _unwatched_runs.pop(address, None) + + +def _runs_in_flight(address: Optional[str]) -> int: + """How many runs `address` has in flight: registered and not yet ended -- + waiting for the card included, paused excluded -- plus the unwatched ones. + + Read from the registry rather than counted up and down, so a run that + crashed, was cancelled or was ended at a restart (runs.fail_orphans) frees + its slot by the state its own terminal event wrote, with nothing to keep + in step. + """ + listed = sum( + 1 for entry in runs.active(limit=sys.maxsize) + if entry.get("client") == address + and entry.get("state") in (runs.STATE_PENDING, runs.STATE_RUNNING) + ) + with _unwatched_lock: + return listed + _unwatched_runs.get(address, 0) + + +def _refuse_past_the_client_cap(request: Request, address: Optional[str]) -> None: + """429 once `address` has `MAX_RUNS_PER_CLIENT` runs in flight. + + Every workstation holds the same API token, so admission alone cannot tell + six clinicians from one client sending sixty runs -- and the other five + would wait behind all sixty. A battery an operator started is exempt: it is + the server measuring itself, several runs at a time on purpose. + """ + cap = settings.MAX_RUNS_PER_CLIENT + if cap <= 0 or benchmark_jobs.is_battery(request.headers.get(benchmark_jobs.KEY_HEADER)): + return + in_flight = _runs_in_flight(address) + if in_flight >= cap: + logger.warning("endpoint=/run status=429 client=%s in_flight=%d cap=%d", + address, in_flight, cap) + raise HTTPException( + status_code=status.HTTP_429_TOO_MANY_REQUESTS, + detail=(f"This workstation already has {in_flight} run(s) in flight, the " + f"most one workstation may have at once ({cap}). Send this one " + "again once one of them has finished."), + ) + + def _client_address(request: Request) -> Optional[str]: """Which workstation asked for this run. @@ -2845,17 +3052,21 @@ async def _detached_run(tool_name: str, request: Request, run_id: str) -> None: result=_collectable(response)) logger.info("endpoint=/run/%s paused (detached)", tool_name) return - runs.finish(run_id, runs.PHASE_DONE, result=_collectable(response)) + # Every ending, and the diagnosis of a failure, reads the run's whole + # event file back: off the event loop (see `_off_loop`). + await _off_loop(functools.partial( + runs.finish, run_id, runs.PHASE_DONE, result=_collectable(response))) except dispatch.RunCancelled: - runs.finish(run_id, runs.PHASE_CANCELLED) + await _off_loop(runs.finish, run_id, runs.PHASE_CANCELLED) except BaseException as exc: # noqa: BLE001 - nobody is left to raise to - diagnosis = _run_diagnosis(exc, run_id, tool_name) + diagnosis = await _off_loop(_run_diagnosis, exc, run_id, tool_name) # Redacted, like every line about a failure: the exception's own text # is a tool's words and may name the file it died on. logger.warning("endpoint=/run/%s detached failure: %s", tool_name, _described(diagnosis)) - runs.finish(run_id, runs.PHASE_FAILED, message=_failure_message(exc), - failure=diagnosis) + await _off_loop(functools.partial( + runs.finish, run_id, runs.PHASE_FAILED, message=_failure_message(exc), + failure=diagnosis)) finally: runs.CURRENT_RUN.reset(token) try: @@ -2881,7 +3092,7 @@ async def _detach(tool_name: str, request: Request, run_id, background_tasks): ) form = await request.form() if any(isinstance(value, StarletteUploadFile) for _, value in form.multi_items()): - runs.discard(run_id) + await _off_loop(runs.discard, run_id) raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail=( @@ -3110,8 +3321,10 @@ async def _tracked_run(run_id: str, label: str, background_tasks: BackgroundTask try: response = await call() except dispatch.RunCancelled: - runs.finish(run_id, runs.PHASE_CANCELLED) - runs.discard(run_id) + # Off the event loop, like every ending: `finish` reads the run's + # whole event file back for the ledger (see `_off_loop`). + await _off_loop(runs.finish, run_id, runs.PHASE_CANCELLED) + await _off_loop(runs.discard, run_id) logger.info("endpoint=%s status=%d", label, CLIENT_CLOSED_REQUEST) raise HTTPException( status_code=CLIENT_CLOSED_REQUEST, detail="Run cancelled by the client." @@ -3120,9 +3333,10 @@ async def _tracked_run(run_id: str, label: str, background_tasks: BackgroundTask # Every failure path, the 404 for an unknown tool included -- the tool # is resolved after the run is registered, so that one now has a # directory to clean up like any other. - runs.finish(run_id, runs.PHASE_FAILED, - failure=_run_diagnosis(exc, run_id)) - runs.discard(run_id) + diagnosis = await _off_loop(_run_diagnosis, exc, run_id) + await _off_loop(functools.partial( + runs.finish, run_id, runs.PHASE_FAILED, failure=diagnosis)) + await _off_loop(runs.discard, run_id) raise finally: runs.CURRENT_RUN.reset(token) @@ -3137,7 +3351,7 @@ async def _tracked_run(run_id: str, label: str, background_tasks: BackgroundTask logger.info("endpoint=%s paused", label) return response - runs.finish(run_id, runs.PHASE_DONE) + await _off_loop(runs.finish, run_id, runs.PHASE_DONE) # Queued rather than done now, so the directory survives until the response # has finished streaming -- which gives a watcher the whole download to # collect the terminal event. Waiting out the TTL instead is not an option: @@ -3338,11 +3552,17 @@ async def run_tool(tool_name: str, request: Request, background_tasks: Backgroun directory down afterwards. A client that sends no `X-Run-Id` takes the first branch and reaches byte-for-byte the behaviour it always had. """ - run_id = _registered_run(request, tool_name) - if request.headers.get(_RUN_DELIVERY_HEADER, "").lower() == _RUN_DETACHED: + # Before a byte of the form is read. Shielded (`_off_loop`), so a + # cancellation cannot land between a slot being taken and the `with` + # below that gives it back: nothing awaits in between. + address = _client_address(request) + detached = request.headers.get(_RUN_DELIVERY_HEADER, "").lower() == _RUN_DETACHED + run_id = await _off_loop(_claimed_run, request, tool_name, address, detached) + if detached: return await _detach(tool_name, request, run_id, background_tasks) if run_id is None: - return await _run_tool(tool_name, request, background_tasks) + with _unwatched_run(address): + return await _run_tool(tool_name, request, background_tasks) # The run id is read by dispatch in the worker thread anyio copies this # context into. It is request scope, not tool input, so it travels the way @@ -3494,6 +3714,7 @@ async def _run_tool(tool_name: str, request: Request, background_tasks: Backgrou _reject_upload_for_unknown_argument(tool, field_name) spec = tool.arguments.get(field_name) _reject_upload_for_scalar(spec, field_name) + _reject_upload_for_model(field_name) extension = _checked_extension(tool, field_name, upload.filename or "") input_path = _staged_input_path(work_dir, field_name, upload.filename or "", extension) @@ -3526,6 +3747,7 @@ async def _run_tool(tool_name: str, request: Request, background_tasks: Backgrou _reject_upload_for_unknown_argument(tool, field_name) spec = tool.arguments.get(field_name) _reject_upload_for_scalar(spec, field_name) + _reject_upload_for_model(field_name) try: session = await anyio.to_thread.run_sync(transfer.get_upload, upload_id) extension = _checked_extension(tool, field_name, session.filename) @@ -3564,8 +3786,7 @@ async def _run_tool(tool_name: str, request: Request, background_tasks: Backgrou # work dir goes now. So do the unclaimed sessions: the reaper would get # them eventually, but that is a TTL's worth of confidential imaging # sitting on disk. - if work_dir: - shutil.rmtree(work_dir, ignore_errors=True) + await _off_loop(_discard, work_dir, ()) for upload_id in upload_references.values(): await anyio.to_thread.run_sync(transfer.discard_upload, upload_id) raise @@ -3590,8 +3811,7 @@ async def _run_tool(tool_name: str, request: Request, background_tasks: Backgrou *( (scope,) if scope else () ), ) except DataNotFoundError as exc: - if work_dir: - shutil.rmtree(work_dir, ignore_errors=True) + await _off_loop(_discard, work_dir, ()) raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=str(exc)) resolved_files.append(resolved) @@ -3612,7 +3832,7 @@ async def _run_tool(tool_name: str, request: Request, background_tasks: Backgrou functools.partial(_extract_folder_argument, spec, path, work_dir, field_name) ) except HTTPException: - shutil.rmtree(work_dir, ignore_errors=True) + await _off_loop(_discard, work_dir, ()) raise args[field_name] = ResolvedPath(path, kind) @@ -3643,7 +3863,7 @@ async def _run_tool(tool_name: str, request: Request, background_tasks: Backgrou # and DATA_DIR is read-only. result = await _run_tool_call(functools.partial(tool.invoke, args)) except ToolArgumentError as exc: - _discard(work_dir, scratch_dirs) + await _off_loop(_discard, work_dir, scratch_dirs) raise HTTPException(status_code=status.HTTP_422_UNPROCESSABLE_CONTENT, detail=str(exc)) except dispatch.ToolFailure as exc: # The tool raised and named its exception class. There is no shared @@ -3663,7 +3883,7 @@ async def _run_tool(tool_name: str, request: Request, background_tasks: Backgrou # broke and why survive that; the name does not. logger.error("endpoint=/run/%s failure: %s", tool_name, _described(_diagnosis(exc, tool_name))) - _discard(work_dir, scratch_dirs) + await _off_loop(_discard, work_dir, scratch_dirs) raise HTTPException( status_code=code, detail=exc.message if code != 500 else "Tool execution failed.", @@ -3673,14 +3893,14 @@ async def _run_tool(tool_name: str, request: Request, background_tasks: Backgrou # image does not carry). 501 rather than 500: nothing the caller changes # will help, and the reason names a missing package, never a path. logger.warning("endpoint=/run/%s status=501", tool_name) - _discard(work_dir, scratch_dirs) + await _off_loop(_discard, work_dir, scratch_dirs) raise HTTPException(status_code=status.HTTP_501_NOT_IMPLEMENTED, detail=str(exc)) except dispatch.RunCancelled: # The client withdrew this run. Cleaned up exactly like a failure -- # nothing will be streamed and the inputs are confidential -- but # re-raised unchanged, because only the caller of this function knows # the run id and so only it can record how the run ended. - _discard(work_dir, scratch_dirs) + await _off_loop(_discard, work_dir, scratch_dirs) raise except Exception as exc: # Not logger.exception: a ToolExecutionError's message is the tool's @@ -3690,19 +3910,11 @@ async def _run_tool(tool_name: str, request: Request, background_tasks: Backgrou logger.error("endpoint=/run/%s status=500 failure: %s\n%s", tool_name, _described(_diagnosis(exc, tool_name)), "".join(traceback.format_tb(exc.__traceback__)).rstrip()) - _discard(work_dir, scratch_dirs) + await _off_loop(_discard, work_dir, scratch_dirs) raise HTTPException(status_code=500, detail="Tool execution failed.") finally: # Uploaded inputs are never needed again past this point. - for input_path in input_paths: - if os.path.exists(input_path): - os.remove(input_path) - # Server-side data (DATA_DIR) is persistent and must never be - # deleted; only backend-materialized temp copies are (see - # ResolvedFile.is_temporary in data_store.py). - for resolved in resolved_files: - if resolved.is_temporary and os.path.exists(resolved.path): - os.remove(resolved.path) + await _off_loop(_remove_staged, input_paths, resolved_files) return await _finish_run( tool, tool_name, start_time, background_tasks, detached, work_dir, @@ -3872,7 +4084,7 @@ async def _finish_run(tool, tool_name: str, start_time: float, logger.exception("endpoint=/run/%s status=500 (packing output)", tool_name) # Never the roots of what the tool NAMED: those are exactly what a # rejected output pointed at -- another run's directory, say. - _discard(work_dir, list(scratch_dirs) + list(confine_to)) + await _off_loop(_discard, work_dir, list(scratch_dirs) + list(confine_to)) raise HTTPException(status_code=500, detail="Tool execution failed.") media_type = _media_type_of(str(result)) diff --git a/server/registry/conventions.py b/server/registry/conventions.py index cc89298..1acf1e0 100644 --- a/server/registry/conventions.py +++ b/server/registry/conventions.py @@ -152,6 +152,19 @@ def is_model(argument_name: str) -> bool: return any(argument_name == name or argument_name.endswith("_" + name) for name in MODEL_NAMES) +# Of MODEL_NAMES, the one that means WEIGHTS: `model` and `*_model`. Never +# uploaded, whatever a deployment opted them out of (main.py +# `_reject_upload_for_model`): a model loader unpickles what it is handed, so an +# uploaded "model" is uploaded code. `*_reference` is not in it -- a reference +# is a scan or a mesh, FlexReg's other timepoint or AutoMatrix's output grid, +# and is the clinician's own to send. +WEIGHTS_NAME = "model" + + +def names_weights(argument_name: str) -> bool: + return argument_name == WEIGHTS_NAME or argument_name.endswith("_" + WEIGHTS_NAME) + + # --- splitting a cohort ---------------------------------------------------- @@ -276,12 +289,19 @@ def derive(arguments: dict, declared: ToolDeployment, batch_defaults: tuple, selectable.update(declared.server_selectable) # "none" is a removal, not a kind: it is how a deployment opts an argument # out of a convention its NAME would otherwise put it in. + opted_out = [] for name, kind in list(selectable.items()): if kind == deployment_module.SERVER_SELECTABLE_NONE: del selectable[name] + opted_out.append(name) hidden = {name for name in arguments if name in TECHNICAL} hidden.update(declared.hidden) + # A model opted out is one the tool finds for itself, and left a plain + # `path` it would otherwise be an upload row on every panel -- for a file + # the server refuses (see WEIGHTS_NAME). Nobody is asked; the tool's own + # default applies, which is what opting it out was for. + hidden.update(name for name in opted_out if names_weights(name)) return replace( declared, diff --git a/server/telemetry.py b/server/telemetry.py index 677db67..d5d7fae 100644 --- a/server/telemetry.py +++ b/server/telemetry.py @@ -176,6 +176,29 @@ def _tree_bytes(path: str, budget: int = 40000) -> Optional[int]: return total +# How long a tree's figure is reused. The panel polls every two seconds, and +# each poll walked TEMP_DIR and DATA_DIR -- up to the budget's 40 000 files +# each -- to say how much they hold: a number that moves on the scale of an +# upload, not of a poll. Half a minute stale is invisible on a gauge. +TREE_BYTES_SECONDS = 30.0 +# path -> (when it was walked, on the monotonic clock; what it held). +_tree_figures: dict = {} + + +def _recent_tree_bytes(path: str) -> Optional[int]: + """`_tree_bytes(path)`, walked at most once every TREE_BYTES_SECONDS.""" + now = time.monotonic() + cached = _tree_figures.get(path) + if cached is not None and now - cached[0] < TREE_BYTES_SECONDS: + return cached[1] + figure = _tree_bytes(path) + # A directory that is not there costs one stat to say so, and must read + # as present the moment it is. + if figure is not None: + _tree_figures[path] = (now, figure) + return figure + + def disk(paths: dict) -> dict: """Free space on the filesystem holding each path, and what it holds. @@ -192,7 +215,7 @@ def disk(paths: dict) -> dict: except OSError: entry["free"] = None entry["total"] = None - entry["used_here"] = _tree_bytes(path) + entry["used_here"] = _recent_tree_bytes(path) report[label] = entry return report @@ -591,7 +614,11 @@ def record_run_end(run_id: str, outcome: str, phase: str = "", pass -def run_ledger(limit: int = 80, tool: Optional[str] = None) -> list: +# How many of the newest runs the operator page lists by default. +LEDGER_PAGE = 80 + + +def run_ledger(limit: int = LEDGER_PAGE, tool: Optional[str] = None) -> list: """The runs this process has seen, newest first, optionally of one tool.""" with _ledger_lock: records = list(_ledger.values()) @@ -844,5 +871,6 @@ def reset() -> None: _ledger.clear() with _trace_lock: _trace.clear() + _tree_figures.clear() global _trace_cpu_previous _trace_cpu_previous = None diff --git a/server/tests/test_compose_logging.py b/server/tests/test_compose_logging.py new file mode 100644 index 0000000..3ae221f --- /dev/null +++ b/server/tests/test_compose_logging.py @@ -0,0 +1,35 @@ +"""Every service that runs the server keeps a bounded log. + +Docker's default `json-file` driver keeps everything a container ever wrote, +for as long as the container exists: a server that runs for months with a +uvicorn access line per poll fills the disk TEMP_DIR shares, and a run that +cannot write its upload fails for a reason nobody looks for in a log. + +Skipped where docker-compose.yml is not beside this checkout: the docker +`test` service mounts server/ alone. +""" + +import os + +import pytest + +yaml = pytest.importorskip("yaml") + +REPO_ROOT = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) +COMPOSE = os.path.join(REPO_ROOT, "docker-compose.yml") + + +@pytest.mark.skipif(not os.path.isfile(COMPOSE), reason="docker-compose.yml is not beside server/") +def test_every_server_service_bounds_its_log(): + with open(COMPOSE, encoding="utf-8") as handle: + services = yaml.safe_load(handle)["services"] + servers = { + name: service for name, service in services.items() + # The test services run pytest, not a server, and are gone when it ends. + if "serve.py" in str(service.get("command")) and "test" not in (service.get("profiles") or ()) + } + assert {"inference", "inference-cpu", "inference-venvs"} <= set(servers) + for name, service in servers.items(): + options = (service.get("logging") or {}).get("options") or {} + assert options.get("max-size"), f"{name} keeps an unbounded log" + assert options.get("max-file"), f"{name} keeps an unbounded log" diff --git a/server/tests/test_dashboard_data.py b/server/tests/test_dashboard_data.py index a4de133..7f1871b 100644 --- a/server/tests/test_dashboard_data.py +++ b/server/tests/test_dashboard_data.py @@ -10,6 +10,7 @@ import json import os +import threading import time import pytest @@ -203,6 +204,59 @@ def test_server_debug_json_carries_the_trace_and_each_runs_chain(): assert all("chain" not in r for r in status["runs"]) +def test_one_panel_poll_lists_the_runs_once_and_reads_the_ledger_once(monkeypatch): + """The panel used to list every run twice (once through `/status`, whose + masked copy it then threw away) and copy the whole ledger twice.""" + import main + runs.register("twice-run-0000000000000001", tool="AREG") + runs.append("twice-run-0000000000000001", runs.PHASE_RUNNING) + for index in range(90): + _finish(f"ledger-run-{index:015d}") + here = threading.get_ident() + calls = {"active": 0, "ledger": 0} + + def counting(name, original): + def counted(*args, **kwargs): + if threading.get_ident() == here: + calls[name] += 1 + return original(*args, **kwargs) + return counted + + monkeypatch.setattr(runs, "active", counting("active", runs.active)) + monkeypatch.setattr(telemetry, "run_ledger", counting("ledger", telemetry.run_ledger)) + report = main.server_debug_data() + assert calls == {"active": 1, "ledger": 1} + # What the page reads is unchanged: the real ids with their chain, and the + # newest 80 finished runs. + entry = [r for r in report["runs"] if r["run_id"] == "twice-run-0000000000000001"][0] + assert entry["chain"] == [] + assert len(report["ledger"]) == 80 + assert report["ledger"][0]["run_id"] == "ledger-run-000000000000089" + + +def test_the_card_is_read_once_per_trace_interval_not_once_per_poll(monkeypatch): + """Each `/status` and each panel poll exec'd `nvidia-smi`: one process per + page per two seconds, on the machine whose card the runs are waiting for.""" + import main + import resources + monkeypatch.setattr(main, "_card_reading", None, raising=False) + # The budget reads the card once per process, at its first use; resolved + # before counting, since that one read is not a poll's. + resources.allocation() + here, calls = threading.get_ident(), [] + + def detect(): + if threading.get_ident() == here: + calls.append(1) + return (48 * 1024 ** 3, 40 * 1024 ** 3) + + monkeypatch.setattr(resources, "detect_vram_bytes", detect) + for _ in range(3): + card = main.server_status()["card"] + assert card == {"free_bytes": 40 * 1024 ** 3, "total_bytes": 48 * 1024 ** 3} + assert len(calls) <= 1 + + def test_a_reaped_run_is_still_drawn_from_the_ledger(): _finish("reaped-run-000000000000001", timeline={ "spans": [{"phase": "running", "start": 1, "end": 9}], "nested": [], "measured": None}) diff --git a/server/tests/test_detached.py b/server/tests/test_detached.py index a44a763..f18f197 100644 --- a/server/tests/test_detached.py +++ b/server/tests/test_detached.py @@ -157,6 +157,40 @@ def test_a_failing_detached_run_says_so_without_naming_a_path(detached_client): runs.discard(run_id) +def test_a_result_that_cannot_be_parked_fails_the_run_as_such(detached_client, monkeypatch): + """Nothing answers a detached run, so a result that cannot be stored for + collection is a failed run -- with the error written for it, not the + NameError an unimported exception class turned it into.""" + import file_utils + import registry + import telemetry + from base import Tool + + class _Producing(Tool): + name = "Unparkable_Probe" + arguments = {} + output_kind = "file" + + def run(self): + path = os.path.join(file_utils.make_scratch_dir(), "result.nii.gz") + with open(path, "wb") as handle: + handle.write(b"volume") + return path + + def full(*_args, **_kwargs): + raise OSError("no space left on device") + + monkeypatch.setitem(registry.TOOLS, _Producing.name, _Producing()) + monkeypatch.setattr(main.transfer, "store_result", full) + run_id = _new_id() + response = detached_client.post( + f"/run/{_Producing.name}", headers=dict(AUTH, **DETACHED, **{"X-Run-Id": run_id})) + assert response.status_code == 202 + assert _terminal(run_id)["state"] == "failed" + assert telemetry.ledger_record(run_id)["failure"]["error_type"] == "ToolExecutionError" + runs.discard(run_id) + + def test_an_unknown_tool_is_refused_before_any_run_is_recorded(detached_client): """The name comes from the URL. Recorded, it reached the status page as whatever string the client sent -- markup included.""" diff --git a/server/tests/test_file_arguments.py b/server/tests/test_file_arguments.py index 30c08d2..7d2fa14 100644 --- a/server/tests/test_file_arguments.py +++ b/server/tests/test_file_arguments.py @@ -1,9 +1,17 @@ """A file argument arrives as bytes, never as a path on the server.""" +import io +import json +import os + +import pytest from fastapi.testclient import TestClient +import registry +from base import ArgSpec, PATH_TYPE, Tool from config import settings from main import app +from registry import conventions, deployment client = TestClient(app) AUTH = {"Authorization": f"Bearer {settings.API_TOKEN}"} @@ -25,3 +33,92 @@ def test_the_forms_empty_placeholder_beside_an_upload_still_works(): data={**FORM, "input": ""}, files={"input": ("input.csv", b"a,b\n1,2\n")}) assert response.status_code == 200, response.text + + +# --------------------------------------------------------------------------- +# A model is never uploaded +# --------------------------------------------------------------------------- +# +# A deployment opts a model argument out of the hosted-name convention with +# "none" when the tool finds its own weights (deployment.toml does it for 21 of +# them). That leaves the argument a plain `path` -- an uploadable file -- and a +# model loader that unpickles what it is given then runs whatever was uploaded. + +class _OptedOutModelTool(Tool): + """What `server_selectable = { seg_model = "none", t1_reference = "none" }` + makes of a tool: both arguments plain file paths.""" + + name = "Opted_Out_Model_Probe" + arguments = { + "scan": ArgSpec(type=PATH_TYPE, required=False), + "seg_model": ArgSpec(type=PATH_TYPE, required=False), + "t1_reference": ArgSpec(type=PATH_TYPE, required=False), + } + output_kind = "text" + ran = [] + + def run(self, scan=None, seg_model=None, t1_reference=None): + _OptedOutModelTool.ran.append({"seg_model": seg_model, "t1_reference": t1_reference}) + return "ran" + + +@pytest.fixture +def opted_out(monkeypatch): + _OptedOutModelTool.ran = [] + monkeypatch.setitem(registry.TOOLS, _OptedOutModelTool.name, _OptedOutModelTool()) + return f"/run/{_OptedOutModelTool.name}" + + +def _staged(): + """What the requests have left in TEMP_DIR, beside the long-lived roots.""" + return set(os.listdir(settings.TEMP_DIR)) - {"runs", "uploads", "results"} + + +def test_an_uploaded_model_is_refused_and_nothing_is_staged(opted_out): + before = _staged() + response = client.post(opted_out, headers=AUTH, + files={"seg_model": ("weights.pth", io.BytesIO(b"\x80\x04 pickle"))}) + assert response.status_code == 422 + assert response.json()["detail"] == ( + "'seg_model' is a model: the server uses its own weights, they cannot be uploaded.") + assert _OptedOutModelTool.ran == [] + assert _staged() == before + + +def test_a_model_sent_through_a_chunked_upload_is_refused_and_the_upload_dropped(opted_out): + payload = b"\x80\x04 pickle" + session = client.post("/uploads", headers=AUTH, + json={"filename": "weights.pth", "size": len(payload)}).json() + client.put(f"/uploads/{session['upload_id']}/parts/0", headers=AUTH, content=payload) + response = client.post(opted_out, headers=AUTH, + data={"__uploads__": json.dumps({"seg_model": session["upload_id"]})}) + assert response.status_code == 422 + assert "is a model" in response.json()["detail"] + assert _OptedOutModelTool.ran == [] + # A refused session does not linger on disk. + assert client.get(f"/uploads/{session['upload_id']}", headers=AUTH).status_code == 404 + + +def test_a_reference_is_still_the_clinician_s_own_upload(opted_out): + """`*_reference` is a scan or a mesh -- FlexReg's other timepoint, + AutoMatrix's output grid -- and is the caller's to send.""" + response = client.post(opted_out, headers=AUTH, + files={"t1_reference": ("baseline.nii", io.BytesIO(b"volume"))}) + assert response.status_code == 200, response.text + assert _OptedOutModelTool.ran and _OptedOutModelTool.ran[0]["t1_reference"] + + +def test_a_model_that_was_opted_out_is_not_offered_to_a_panel(): + """A panel renders an upload row for every file argument it is shown; the + server would now refuse what that row sends. Hidden, nobody is asked.""" + arguments = {"seg_model": {"type": "path"}, "t1_reference": {"type": "path"}, + "scan": {"type": "path"}} + declared = deployment.ToolDeployment(server_selectable={ + "seg_model": deployment.SERVER_SELECTABLE_NONE, + "t1_reference": deployment.SERVER_SELECTABLE_NONE, + }) + merged = conventions.derive(arguments, declared, + (settings.BATCH_MAX_MB, settings.BATCH_MAX_FILES)) + assert "seg_model" in merged.hidden + assert "t1_reference" not in merged.hidden + assert "scan" not in merged.hidden diff --git a/server/tests/test_main.py b/server/tests/test_main.py index 4a2c209..8218db0 100644 --- a/server/tests/test_main.py +++ b/server/tests/test_main.py @@ -8,7 +8,9 @@ import io import json import os +import secrets import shutil +import threading import zipfile # Set before importing main, so config.Settings() picks up a known token @@ -1786,6 +1788,88 @@ def run(self, scan): assert seen == ["worker thread"] +def _calls_on_the_loop(monkeypatch, *targets): + """Record every call to `targets` -- (owner, attribute) pairs -- made on + the event loop that serves this test's request, the thread `_run_tool` + runs on, and on no other thread.""" + import main + loop_thread, seen = [], [] + real_run_tool = main._run_tool + + async def run_tool(*args, **kwargs): + loop_thread.append(threading.get_ident()) + return await real_run_tool(*args, **kwargs) + + def spy(name, real): + def recorded(*args, **kwargs): + if threading.get_ident() in loop_thread: + seen.append(name) + return real(*args, **kwargs) + return recorded + + monkeypatch.setattr(main, "_run_tool", run_tool) + for owner, attribute in targets: + monkeypatch.setattr(owner, attribute, spy(attribute, getattr(owner, attribute))) + return seen + + +def _deletions_on_the_loop(monkeypatch): + return _calls_on_the_loop(monkeypatch, (shutil, "rmtree"), (os, "remove")) + + +@pytest.mark.parametrize("watched", [False, True], ids=["no run id", "with a run id"]) +def test_a_refused_runs_inputs_are_deleted_off_the_event_loop(monkeypatch, watched): + """Deleting a staged CBCT costs ~105 ms a gigabyte; on the event loop that + froze every other request -- uploads, /status, the panel -- for as long. + A watched run's directory goes the same way, and so does its ending.""" + + class _Probe(Tool): + name = "Delete_Thread_Probe" + arguments = {"fixed_image": ArgSpec(type="file", required=True), + "moving_image": ArgSpec(type="file", required=True)} + output_kind = "text" + + def run(self, fixed_image, moving_image): + return "unused" + + monkeypatch.setitem(registry.TOOLS, _Probe.name, _Probe()) + seen = _deletions_on_the_loop(monkeypatch) + headers = {"Authorization": f"Bearer {TOKEN}"} + if watched: + headers["X-Run-Id"] = secrets.token_urlsafe(24) + response = client.post( + f"/run/{_Probe.name}", + headers=headers, + files={"fixed_image": ("a.nii.gz", gzip.compress(b"a" * 100), "application/gzip")}, + ) + assert response.status_code == 422, response.text + assert seen == [] + + +@pytest.mark.parametrize("arguments, delivery, code", [ + ({"text_1": "hello", "text_2": "world"}, {}, 200), + ({"text_1": "hello"}, {}, 422), + ({"text_1": "hello", "text_2": "world"}, {"X-Run-Delivery": "detached"}, 202), + ({"text_1": "hello"}, {"X-Run-Delivery": "detached"}, 202), +], ids=["done", "failed", "detached done", "detached failed"]) +def test_a_watched_runs_ending_is_read_back_off_the_event_loop(monkeypatch, arguments, + delivery, code): + """`runs.finish` reads the run's whole event file back for the ledger: a + chatty tool's 100k lines are half a second, and on the event loop every + other request waited for it. The diagnosis of a failure reads it too.""" + from wire import runs + seen = _calls_on_the_loop(monkeypatch, (runs.EventReader, "read")) + run_id = secrets.token_urlsafe(24) + response = client.post( + "/run/Test_Tool", + headers={"Authorization": f"Bearer {TOKEN}", "X-Run-Id": run_id, **delivery}, + data=arguments, + ) + assert response.status_code == code, response.text + assert seen == [] + runs.discard(run_id) + + def test_a_truncated_gzip_is_refused_rather_than_segmented(monkeypatch, tmp_path): """ITK's NIfTI reader does not report a truncated gzip: it believes the header's dimensions and zero-fills the rest. Measured on a scan cut to diff --git a/server/tests/test_page_polling.py b/server/tests/test_page_polling.py new file mode 100644 index 0000000..0fce3bb --- /dev/null +++ b/server/tests/test_page_polling.py @@ -0,0 +1,137 @@ +"""The two pages that poll this server ask again only once they have been answered. + +`/` polled `/status` on `setInterval(poll, 2000)`, and the admin panel re-armed +its timer whether or not the last fetch had come back. Against a server that +answers slowly -- which is exactly when somebody opens the page to see why -- +each tab stacked one more request every two seconds behind the ones already +waiting, and made the server slower still. + +Each page's own script is run in node against a DOM that answers everything +and does nothing, a `fetch` that never settles, and fake timers advanced 20 +seconds. Skipped when node is not installed: nothing else here needs it. +""" + +import json +import re +import shutil +import subprocess + +import pytest + +from wire import debug_page, status_page + +NODE = shutil.which("node") +pytestmark = pytest.mark.skipif(NODE is None, reason="node is not installed") + +# A DOM in which every name exists and every call returns another such +# object, so a page script runs top to bottom without a browser; `fetch` +# never settles, or fails at once; the clock is advanced by hand. Prints how +# many fetches of `watched` were started. +_HARNESS = r""" +const vm = require("vm"); +const fs = require("fs"); +const [source, watched, seconds, answer] = [fs.readFileSync(process.argv[2], "utf8"), + process.argv[3], Number(process.argv[4]), process.argv[5]]; +function stub() { + return new Proxy(function () {}, { + get(target, key) { + if (key === Symbol.toPrimitive) { return () => ""; } + if (key === "then" || typeof key === "symbol") { return undefined; } + return stub(); + }, + set() { return true; }, + apply() { return stub(); }, + construct() { return stub(); }, + }); +} +let clock = 0, started = 0; +const timers = new Map(); +let next = 1; +function arm(fn, ms, every) { + const id = next++; + timers.set(id, { at: clock + Math.max(0, Number(ms) || 0), fn, every }); + return id; +} +const own = { + fetch(url) { + if (String(url).indexOf(watched) !== -1) { started += 1; } + if (answer === "refused") { return Promise.reject(new Error("refused")); } + return new Promise(function () {}); + }, + setTimeout: (fn, ms) => arm(fn, ms, null), + setInterval: (fn, ms) => arm(fn, ms, Math.max(1, Number(ms) || 0)), + clearTimeout: (id) => timers.delete(id), + clearInterval: (id) => timers.delete(id), + localStorage: { getItem: () => "a-token", setItem() {}, removeItem() {} }, +}; +const scope = new Proxy({}, { + has: () => true, + get(target, key) { + if (key === Symbol.unscopables) { return undefined; } + if (key in own) { return own[key]; } + if (key === "window" || key === "self" || key === "globalThis") { return scope; } + if (key in globalThis) { return globalThis[key]; } + return stub(); + }, + set(target, key, value) { own[key] = value; return true; }, +}); +const context = vm.createContext({ scope }); +vm.runInContext("with (scope) {\n" + source + "\n}", context); +(async function () { + const flush = () => new Promise((resolve) => setImmediate(resolve)); + await flush(); + const end = seconds * 1000; + for (;;) { + let due = null; + for (const [id, timer] of timers) { + if (timer.at <= end && (due === null || timer.at < timers.get(due).at)) { due = id; } + } + if (due === null) { break; } + const timer = timers.get(due); + clock = timer.at; + if (timer.every) { timer.at += timer.every; } else { timers.delete(due); } + timer.fn(); + await flush(); + } + process.stdout.write(JSON.stringify({ started })); +})(); +""" + + +def _scripts(page: str) -> str: + return "\n;\n".join(re.findall(r"", page, flags=re.S)) + + +def _fetches_started(tmp_path, page: str, watched: str, answer: str = "never", + seconds: int = 20) -> int: + harness = tmp_path / "harness.js" + harness.write_text(_HARNESS) + script = tmp_path / "page.js" + script.write_text(_scripts(page)) + # A guard, not a measurement: a page script that loops for ever must fail + # this test rather than hang the suite. + finished = subprocess.run([NODE, str(harness), str(script), watched, str(seconds), answer], + capture_output=True, text=True, timeout=30) + assert finished.returncode == 0, finished.stderr + return json.loads(finished.stdout)["started"] + + +def test_the_status_page_waits_for_its_answer_before_asking_again(tmp_path): + started = _fetches_started(tmp_path, status_page.STATUS_PAGE, "status") + assert started <= 2 + + +def test_the_admin_panel_waits_for_its_answer_before_asking_again(tmp_path): + started = _fetches_started(tmp_path, debug_page.DEBUG_PAGE, "admin-panel.json") + assert started <= 2 + + +@pytest.mark.parametrize("page, watched", [ + (status_page.STATUS_PAGE, "status"), + (debug_page.DEBUG_PAGE, "admin-panel.json"), +], ids=["status page", "admin panel"]) +def test_a_page_answered_at_once_keeps_polling(tmp_path, page, watched): + """Waiting for the answer must not stop the polling: a refused fetch has + settled, and the next one goes two seconds later.""" + started = _fetches_started(tmp_path, page, watched, answer="refused") + assert started >= 10 diff --git a/server/tests/test_runs.py b/server/tests/test_runs.py index c3c7c70..e517d43 100644 --- a/server/tests/test_runs.py +++ b/server/tests/test_runs.py @@ -223,6 +223,40 @@ def write_the_rest(): ] +def test_a_quiet_stream_says_it_is_still_there(run_id, monkeypatch): + """A tool silent for longer than the client's 30 s read timeout -- a + registration, a model loading -- made the client drop the stream and + reconnect, replaying every event from the start. A comment line keeps the + connection reading as alive and is never an event.""" + import main + monkeypatch.setattr(settings, "RUN_EVENT_POLL_SECONDS", 0.01) + monkeypatch.setattr(main, "_SSE_HEARTBEAT_SECONDS", 0.05, raising=False) + runs.append(run_id, runs.PHASE_RUNNING) + directory = runs.run_directory(run_id) + reads, real = [], runs.EventReader.read + + def read(reader): + # The run ends after this many polls, so the stream ends and the test + # cannot hang on it. + if reader.directory == directory: + reads.append(1) + if len(reads) == 40: + runs.finish(run_id, runs.PHASE_DONE) + return real(reader) + + monkeypatch.setattr(runs.EventReader, "read", read) + comments, data = [], [] + with client.stream("GET", f"/runs/{run_id}/events", headers=AUTH, timeout=20) as response: + assert response.status_code == 200 + for line in response.iter_lines(): + if line.startswith(":"): + comments.append(line) + elif line.startswith("data: "): + data.append(json.loads(line[len("data: "):])) + assert len(comments) >= 1 + assert [event["phase"] for event in data] == [runs.PHASE_RUNNING, runs.PHASE_DONE] + + def test_the_stream_ends_when_the_run_directory_goes_away(run_id): """A run whose request died between the last event and the cleanup. The watcher must stop rather than poll a directory that will never change.""" @@ -1053,3 +1087,246 @@ def test_a_hold_adds_what_it_says_and_does_not_restart_the_idle_clock(run_id, tm before = runs._expires_at(directory) runs.extend_pause(run_id, 600) assert runs._expires_at(directory) == pytest.approx(before + 600, abs=1) + + +# --------------------------------------------------------------------------- +# The live listing does not re-read a run that has not changed +# --------------------------------------------------------------------------- + +def _counting_parses(monkeypatch): + """Count `_normalised` calls made on THIS thread only: a reaper or a + leftover worker of another test parsing its own run must not be counted.""" + calls = [] + here = threading.get_ident() + original = runs._normalised + + def counted(*args, **kwargs): + if threading.get_ident() == here: + calls.append(1) + return original(*args, **kwargs) + + monkeypatch.setattr(runs, "_normalised", counted) + return calls + + +def test_the_listing_reads_an_unchanged_run_once(tmp_path, monkeypatch): + """Every page polls the listing every two seconds; re-reading and + re-verifying every event of every run on each poll cost 2.4 ms a run, two + and a half seconds a panel poll with 500 runs on disk.""" + monkeypatch.setattr(settings, "TEMP_DIR", str(tmp_path)) + made = [runs.register(_new_id(), tool="AMASSS") for _ in range(20)] + for identifier in made: + for _ in range(50): + runs.append(identifier, runs.PHASE_RUNNING, fraction=0.5) + first = runs.active(with_chain=True) + + calls = _counting_parses(monkeypatch) + assert runs.active() == [{k: v for k, v in entry.items() if k != "chain"} for entry in first] + assert runs.active(with_chain=True) == first + assert len(calls) == 0 + + +def test_the_listing_rereads_a_run_that_moved(tmp_path, monkeypatch): + monkeypatch.setattr(settings, "TEMP_DIR", str(tmp_path)) + identifier = runs.register(_new_id(), tool="AMASSS") + runs.append(identifier, runs.PHASE_RUNNING, fraction=0.25) + assert _listed(identifier)["fraction"] == 0.25 + runs.append(identifier, runs.PHASE_RUNNING, fraction=0.75) + assert _listed(identifier)["fraction"] == 0.75 + runs.finish(identifier, runs.PHASE_DONE) + assert _listed(identifier)["state"] == runs.STATE_DONE + + +def test_the_listing_forgets_a_run_that_is_gone(tmp_path, monkeypatch): + monkeypatch.setattr(settings, "TEMP_DIR", str(tmp_path)) + identifier = runs.register(_new_id(), tool="AMASSS") + runs.append(identifier, runs.PHASE_RUNNING) + runs.active() + directory = runs.run_directory(identifier) + assert directory in runs._LISTED + runs.discard(identifier) + runs.active() + assert directory not in runs._LISTED + + +# --------------------------------------------------------------------------- +# A run the previous server left in flight is ended when the next one starts +# --------------------------------------------------------------------------- + +def _states(): + return {entry["run_id"]: entry for entry in runs.active()} + + +def test_a_run_a_previous_server_left_in_flight_is_failed_at_startup(tmp_path, monkeypatch): + """TEMP_DIR outlives a restart, and nothing else would ever end such a run: + the event stream touches its directory every quarter second, so a watched + orphan is never reaped, and the client waits for a terminal event for ever.""" + monkeypatch.setattr(settings, "TEMP_DIR", str(tmp_path)) + running = runs.register(_new_id(), tool="AMASSS") + runs.append(running, runs.PHASE_RUNNING, fraction=0.4) + pending = runs.register(_new_id(), tool="AMASSS") + finished = runs.register(_new_id(), tool="AMASSS") + runs.finish(finished, runs.PHASE_DONE) + idle = os.stat(runs.run_directory(running)).st_mtime_ns + + with TestClient(app): + pass + + # Ending it is not the run being active: it keeps the idle clock it had, + # so an abandoned one is still reaped on time. Checked first, since + # reading the events below is a client's read and does stamp it. + assert os.stat(runs.run_directory(running)).st_mtime_ns == idle + listed = _states() + assert {entry["state"] for entry in listed.values() if entry["run_id"] != finished} \ + == {runs.STATE_FAILED} + assert listed[finished]["state"] == runs.STATE_DONE + assert runs.read_events(running)[-1]["message"] == "The server restarted during this run." + + +def test_a_paused_run_survives_a_restart_paused(tmp_path, monkeypatch): + """A pause waits for a person and holds no machine: a restart must leave + it resumable.""" + monkeypatch.setattr(settings, "TEMP_DIR", str(tmp_path)) + job = tmp_path / "job" + job.mkdir() + paused = runs.register(_new_id(), tool="AREG") + runs.append(paused, runs.PHASE_RUNNING) + runs.pause(paused, str(job), "ASO") + + with TestClient(app): + pass + + assert _states()[paused]["state"] == runs.STATE_PAUSED + assert runs.read_events(paused)[-1]["phase"] == runs.PHASE_PAUSED + assert runs.paused_at(paused) is not None + + +# --------------------------------------------------------------------------- +# The listing reads a live run on from where it stopped +# --------------------------------------------------------------------------- + +def _tool_line(directory, record, newline=True): + """Append a line the way a TOOL does: unsigned, O_APPEND, one write.""" + payload = json.dumps(record) + ("\n" if newline else "") + with open(os.path.join(directory, runs.EVENTS_FILE), "a", encoding="utf-8") as handle: + handle.write(payload) + + +def test_a_listing_reads_only_what_a_live_run_appended(tmp_path, monkeypatch): + """A live run changes between any two listings, and the cap on runs per + workstation lists every run at every POST /run: re-reading the whole file + on each change cost 634 ms for one run with 100k lines.""" + monkeypatch.setattr(settings, "TEMP_DIR", str(tmp_path)) + identifier = runs.register(_new_id(), tool="AMASSS") + for index in range(1500): + runs.append(identifier, runs.PHASE_RUNNING, fraction=index / 2000) + runs.active() + runs.append(identifier, runs.PHASE_RUNNING, fraction=0.9) + + calls = _counting_parses(monkeypatch) + assert _listed(identifier)["fraction"] == 0.9 + assert len(calls) == 1 + + +def test_a_tool_writing_after_a_cancelled_event_leaves_the_run_cancelled(tmp_path, monkeypatch): + monkeypatch.setattr(settings, "TEMP_DIR", str(tmp_path)) + identifier = runs.register(_new_id(), tool="AMASSS") + runs.append(identifier, runs.PHASE_RUNNING) + runs.active() + runs.finish(identifier, runs.PHASE_CANCELLED) + assert _listed(identifier)["state"] == runs.STATE_CANCELLED + _tool_line(runs.run_directory(identifier), {"fraction": 0.9, "message": "still going"}) + assert _listed(identifier)["state"] == runs.STATE_CANCELLED + assert _listed(identifier)["phase"] == runs.PHASE_CANCELLED + + +def test_a_rewritten_events_file_is_read_afresh(tmp_path, monkeypatch): + """Shorter than where the listing had read to: not an append, so nothing + read before it can be trusted.""" + monkeypatch.setattr(settings, "TEMP_DIR", str(tmp_path)) + identifier = runs.register(_new_id(), tool="AMASSS") + for _ in range(5): + runs.append(identifier, runs.PHASE_RUNNING, fraction=0.5) + runs.finish(identifier, runs.PHASE_DONE) + assert _listed(identifier)["state"] == runs.STATE_DONE + path = os.path.join(runs.run_directory(identifier), runs.EVENTS_FILE) + with open(path, "r+", encoding="utf-8") as handle: + handle.truncate(0) + runs.append(identifier, runs.PHASE_QUEUED_GPU) + assert _listed(identifier)["phase"] == runs.PHASE_QUEUED_GPU + + +def _whole_read(identifier): + """What the listing would say reading the file from the start, once.""" + events = runs.EventReader(runs.run_directory(identifier), keep_alive=False).read() + latest = events[-1] if events else None + return { + "state": latest["state"] if latest else runs.STATE_PENDING, + "phase": latest["phase"] if latest else runs.PHASE_RECEIVED, + "fraction": latest["fraction"] if latest else None, + "depth": latest["depth"] if latest else 0, + "chain": runs.timeline(events)["chain"], + } + + +def _as_listed(identifier): + entry = next(e for e in runs.active(with_chain=True) if e["run_id"] == identifier) + return {key: entry[key] for key in ("state", "phase", "fraction", "depth", "chain")} + + +@pytest.mark.parametrize("seed", range(12)) +def test_reading_on_says_what_reading_from_the_start_says(tmp_path, monkeypatch, seed): + """Appends of every kind -- server phases, tool progress, log lines, + nested calls opening and closing at several depths, a line torn mid-write, + and writes after a terminal event -- listed after each one, compared with + a whole read of the same file.""" + import random + + monkeypatch.setattr(settings, "TEMP_DIR", str(tmp_path)) + chance = random.Random(seed) + identifier = runs.register(_new_id(), tool="AREG") + directory = runs.run_directory(identifier) + clock = [1_000_000.0] + open_calls = [] + tools = ("ASO", "ALI_CBCT", "AMASSS", "AREG_CBCT") + + def at(): + clock[0] += chance.uniform(0.01, 2.0) + return clock[0] + + for step in range(80): + roll = chance.random() + if roll < 0.2: + runs.append(identifier, chance.choice( + [runs.PHASE_STAGING, runs.PHASE_QUEUED_GPU, runs.PHASE_RUNNING, + runs.PHASE_PACKAGING]), fraction=chance.random()) + elif roll < 0.45: + _tool_line(directory, {"at": at(), "fraction": chance.random(), + "message": "working", "depth": chance.randint(0, 2)}) + elif roll < 0.55: + _tool_line(directory, {"at": at(), "kind": "log", "level": "info", + "audience": "admin", "message": "a log line"}) + elif roll < 0.75: + depth = len(open_calls) + 1 + tool = chance.choice(tools) + open_calls.append((tool, depth)) + _tool_line(directory, {"at": at(), "depth": depth, "tool": tool, + "call": ".".join(["1"] * depth), "edge": "open"}) + elif roll < 0.9 and open_calls: + tool, depth = open_calls.pop() + _tool_line(directory, {"at": at(), "depth": depth, "tool": tool, + "call": ".".join(["1"] * depth), "edge": "close"}) + elif roll < 0.95: + # A line caught mid-write: the listing must neither parse it in + # half nor lose it once it is complete. + record = json.dumps({"at": at(), "fraction": 0.5, "message": "torn"}) + cut = chance.randint(1, len(record) - 1) + with open(os.path.join(directory, runs.EVENTS_FILE), "a", encoding="utf-8") as handle: + handle.write(record[:cut]) + assert _as_listed(identifier) == _whole_read(identifier), f"step {step} (torn)" + with open(os.path.join(directory, runs.EVENTS_FILE), "a", encoding="utf-8") as handle: + handle.write(record[cut:] + "\n") + elif step > 40: + runs.finish(identifier, chance.choice( + [runs.PHASE_DONE, runs.PHASE_FAILED, runs.PHASE_CANCELLED])) + assert _as_listed(identifier) == _whole_read(identifier), f"step {step}" diff --git a/server/tests/test_runs_per_client.py b/server/tests/test_runs_per_client.py new file mode 100644 index 0000000..46fe018 --- /dev/null +++ b/server/tests/test_runs_per_client.py @@ -0,0 +1,295 @@ +"""One workstation cannot fill the queue for every other one. + +Every workstation holds the same API token, so admission alone cannot tell +"six clinicians at once" from "one client sending sixty": both are sixty runs +waiting for the card, and the five other workstations wait behind them. A run +in flight -- registered and not yet ended -- counts against the address that +sent it, past `MAX_RUNS_PER_CLIENT` the next one is a 429, and a paused run +counts for nothing: it waits for a person and holds no machine. + +The addresses are documentation addresses (RFC 5737). +""" + +import io +import os +import secrets +import threading +import time + +import pytest +from fastapi.testclient import TestClient + +import benchmark_client +import benchmark_jobs +import registry +from base import ArgSpec, Tool +from config import settings +from main import app +from wire import runs + +AUTH = {"Authorization": f"Bearer {settings.API_TOKEN}"} +FIRST, SECOND = "192.0.2.10", "192.0.2.20" + + +def _client(address): + return TestClient(app, client=(address, 50000)) + + +def _new_id() -> str: + return secrets.token_urlsafe(24) + + +@pytest.fixture(autouse=True) +def _own_registry(tmp_path, monkeypatch): + """A registry of this test's own, holding nothing but what it puts there.""" + monkeypatch.setattr(settings, "TEMP_DIR", str(tmp_path)) + monkeypatch.setattr(settings, "MAX_RUNS_PER_CLIENT", 2) + + +def _in_flight(address, count=2, phase=runs.PHASE_RUNNING): + made = [] + for _ in range(count): + identifier = runs.register(_new_id(), tool="AMASSS", client=address) + runs.append(identifier, phase) + made.append(identifier) + return made + + +def _post(address, run_id=None, **kwargs): + headers = dict(AUTH, **({"X-Run-Id": run_id} if run_id else {})) + headers.update(kwargs.pop("headers", {})) + return _client(address).post("/run/Test_Tool", headers=headers, + data={"text_1": "a", "text_2": "b"}, **kwargs) + + +def test_the_run_past_the_cap_is_refused_and_nothing_is_staged(tmp_path): + _in_flight(FIRST) + refused = _new_id() + before = set(os.listdir(tmp_path)) + + response = _post(FIRST, refused, files={"text_3": ("x.nii.gz", io.BytesIO(b"scan"))}) + + assert response.status_code == 429 + assert "2 run(s) in flight" in response.json()["detail"] + # Refused before it was registered: nothing to watch, nothing on disk. + assert _client(FIRST).get(f"/runs/{refused}", headers=AUTH).status_code == 404 + assert set(os.listdir(tmp_path)) == before + + +def test_another_workstation_is_not_held_by_the_first_one(): + _in_flight(FIRST) + assert _post(SECOND, _new_id()).status_code == 200 + + +@pytest.mark.parametrize("ending", [runs.PHASE_DONE, runs.PHASE_FAILED, runs.PHASE_CANCELLED]) +def test_a_run_that_ended_frees_its_slot(ending): + first, _second = _in_flight(FIRST) + assert _post(FIRST, _new_id()).status_code == 429 + runs.finish(first, ending) + assert _post(FIRST, _new_id()).status_code == 200 + + +def test_a_run_waiting_for_the_card_still_counts(): + _in_flight(FIRST, phase=runs.PHASE_QUEUED_GPU) + assert _post(FIRST, _new_id()).status_code == 429 + + +def test_a_paused_run_does_not_count(tmp_path): + first, _second = _in_flight(FIRST) + job = tmp_path / "job" + job.mkdir() + runs.pause(first, str(job), "ASO") + assert _post(FIRST, _new_id()).status_code == 200 + + +def test_zero_is_no_cap(monkeypatch): + monkeypatch.setattr(settings, "MAX_RUNS_PER_CLIENT", 0) + _in_flight(FIRST, count=5) + assert _post(FIRST, _new_id()).status_code == 200 + + +def test_a_run_sent_without_an_id_counts_while_it_runs(monkeypatch): + """No `X-Run-Id`, no registry entry: counted in this process instead, for + exactly as long as the request lasts.""" + monkeypatch.setattr(settings, "MAX_RUNS_PER_CLIENT", 1) + started, release = threading.Event(), threading.Event() + + class _Holding(Tool): + name = "Holding_Probe" + arguments = {"text": ArgSpec(type=str)} + output_kind = "text" + + def run(self, text): + started.set() + # Bounded, so a refusal that never comes cannot hang the suite. + release.wait(timeout=20) + return text + + monkeypatch.setitem(registry.TOOLS, _Holding.name, _Holding()) + answers = [] + with TestClient(app, client=(FIRST, 50000)) as shared: + holder = threading.Thread(target=lambda: answers.append(shared.post( + f"/run/{_Holding.name}", headers=AUTH, data={"text": "x"}))) + holder.start() + try: + assert started.wait(timeout=20) + refused = shared.post("/run/Test_Tool", headers=AUTH, + data={"text_1": "a", "text_2": "b"}) + finally: + release.set() + holder.join(timeout=20) + assert refused.status_code == 429 + assert answers and answers[0].status_code == 200 + # And the slot is free again once that request has ended. + assert shared.post("/run/Test_Tool", headers=AUTH, + data={"text_1": "a", "text_2": "b"}).status_code == 200 + + +class _Running: + def poll(self): + return None + + +def test_a_battery_launched_by_an_operator_is_not_capped(monkeypatch): + """A benchmark battery runs from this server's own loopback, several runs + at a time, because measuring concurrency is what it is for.""" + key = secrets.token_urlsafe(32) + monkeypatch.setattr(benchmark_jobs, "_current", {"_key": key, "_process": _Running()}) + _in_flight(FIRST) + battery = {benchmark_jobs.KEY_HEADER: key} + assert _post(FIRST, _new_id(), headers=battery).status_code == 200 + # A key of its own making buys nothing. + assert _post(FIRST, _new_id(), headers={benchmark_jobs.KEY_HEADER: "guessed"}).status_code == 429 + + +def test_a_finished_battery_s_key_buys_nothing(monkeypatch): + class _Ended: + def poll(self): + return 0 + + key = secrets.token_urlsafe(32) + monkeypatch.setattr(benchmark_jobs, "_current", {"_key": key, "_process": _Ended()}) + _in_flight(FIRST) + assert _post(FIRST, _new_id(), headers={benchmark_jobs.KEY_HEADER: key}).status_code == 429 + + +def test_the_battery_and_the_server_agree_on_the_key_they_share(): + """Two files that cannot import each other: the battery runs as a script.""" + assert benchmark_client.BENCHMARK_KEY_HEADER == benchmark_jobs.KEY_HEADER + assert benchmark_client.BENCHMARK_KEY_ENV == benchmark_jobs.KEY_ENV + + +def test_the_count_is_taken_off_the_event_loop(monkeypatch): + """Counting lists every run on disk; on the event loop that held every + other request -- uploads, the event streams, /status -- for as long.""" + import main + + loop_thread, listed_on = [], [] + real_run_tool, real_active = main._run_tool, runs.active + + async def run_tool(*args, **kwargs): + loop_thread.append(threading.get_ident()) + return await real_run_tool(*args, **kwargs) + + def active(*args, **kwargs): + listed_on.append(threading.get_ident()) + return real_active(*args, **kwargs) + + monkeypatch.setattr(main, "_run_tool", run_tool) + monkeypatch.setattr(runs, "active", active) + assert _post(FIRST, _new_id()).status_code == 200 + assert loop_thread and listed_on + assert [thread for thread in listed_on if thread in loop_thread] == [] + + +def test_two_runs_counted_at_the_same_instant_cannot_both_pass_the_cap(monkeypatch): + """The count and the registration are one step: two requests arriving + together must not both count the same free slot and both register.""" + import main + + monkeypatch.setattr(settings, "MAX_RUNS_PER_CLIENT", 1) + together = threading.Barrier(2) + real_count = main._runs_in_flight + + def counted_together(address): + # Holds each count back until the other has counted too, between the + # count and the registration -- so two counts that CAN overlap both + # see the slot free. Times out, and goes on, when the second is held + # back, as it must be. + counted = real_count(address) + try: + together.wait(timeout=1) + except threading.BrokenBarrierError: + pass + return counted + + release = threading.Event() + + class _Holding(Tool): + name = "Holding_Race_Probe" + arguments = {"text": ArgSpec(type=str)} + output_kind = "text" + + def run(self, text): + release.wait(timeout=20) + return text + + monkeypatch.setattr(main, "_runs_in_flight", counted_together) + monkeypatch.setitem(registry.TOOLS, _Holding.name, _Holding()) + answers = [] + with TestClient(app, client=(FIRST, 50000)) as shared: + def post(): + answers.append(shared.post(f"/run/{_Holding.name}", + headers=dict(AUTH, **{"X-Run-Id": _new_id()}), + data={"text": "x"}).status_code) + + senders = [threading.Thread(target=post) for _ in range(2)] + for sender in senders: + sender.start() + deadline = time.monotonic() + 10 + while not answers and time.monotonic() < deadline: + time.sleep(0.02) + release.set() + for sender in senders: + sender.join(timeout=20) + assert sorted(answers) == [200, 429] + + +@pytest.mark.parametrize("with_id", [False, True], ids=["unwatched", "watched"]) +def test_a_run_cancelled_while_it_claims_its_slot_keeps_none(monkeypatch, with_id): + """A client that hangs up while its count is being taken: the claim runs on + in its thread, so it must give the slot back rather than hold it forever.""" + import anyio + from fastapi import BackgroundTasks + from starlette.requests import Request + + import main + + monkeypatch.setattr(settings, "MAX_RUNS_PER_CLIENT", 1) + address = "192.0.2.30" + real_claim = main._claimed_run + + def slow_claim(*args, **kwargs): + time.sleep(0.3) + return real_claim(*args, **kwargs) + + monkeypatch.setattr(main, "_claimed_run", slow_claim) + headers = [(b"authorization", AUTH["Authorization"].encode())] + if with_id: + headers.append((b"x-run-id", _new_id().encode())) + scope = {"type": "http", "method": "POST", "path": "/run/Test_Tool", "headers": headers, + "query_string": b"", "client": (address, 50000), "server": ("testserver", 80), + "scheme": "http", "app": app} + + async def never_sends(): + await anyio.sleep(30) + return {"type": "http.disconnect"} + + async def hang_up_early(): + with anyio.move_on_after(0.1): + await main.run_tool("Test_Tool", Request(scope, never_sends), BackgroundTasks()) + await anyio.sleep(0.5) # the shielded claim finishing in its thread + + anyio.run(hang_up_early) + assert main._unwatched_runs.get(address) is None + assert main._runs_in_flight(address) == 0 diff --git a/server/tests/test_telemetry.py b/server/tests/test_telemetry.py index f6d128e..7fcbc13 100644 --- a/server/tests/test_telemetry.py +++ b/server/tests/test_telemetry.py @@ -90,6 +90,46 @@ def test_disk_counts_what_is_under_the_path(tmp_path): assert report["here"]["total"] > 0 +def _stats_under(monkeypatch, root): + """Count `os.stat` calls on paths under `root` -- not `root` itself, which + the report still checks exists -- made on this thread only.""" + import threading + calls, here, real = [], threading.get_ident(), os.stat + + def counted(path, *args, **kwargs): + if threading.get_ident() == here and str(path).startswith(str(root) + os.sep): + calls.append(path) + return real(path, *args, **kwargs) + + monkeypatch.setattr(os, "stat", counted) + return calls + + +def test_a_tree_is_walked_once_per_half_minute_not_once_per_poll(tmp_path, monkeypatch): + """The panel polls every two seconds, and each poll walked TEMP_DIR and + DATA_DIR -- up to 40 000 files each -- to say how much they held.""" + tree = tmp_path / "tree" + tree.mkdir() + for index in range(200): + (tree / f"blob-{index}").write_bytes(b"x" * 100) + first = telemetry.disk({"here": str(tree)})["here"]["used_here"] + + calls = _stats_under(monkeypatch, tree) + assert telemetry.disk({"here": str(tree)})["here"]["used_here"] == first + assert len(calls) == 0 + + +def test_a_tree_is_walked_again_once_its_figure_is_stale(tmp_path, monkeypatch): + tree = tmp_path / "tree" + tree.mkdir() + (tree / "blob").write_bytes(b"x" * 20000) + before = telemetry.disk({"here": str(tree)})["here"]["used_here"] + (tree / "another").write_bytes(b"x" * 20000) + later = time.monotonic() + telemetry.TREE_BYTES_SECONDS + 1 + monkeypatch.setattr(telemetry.time, "monotonic", lambda: later) + assert telemetry.disk({"here": str(tree)})["here"]["used_here"] >= before + 20000 + + def test_in_flight_requests_count_up_and_back_down(): telemetry.request_started() telemetry.request_started() diff --git a/server/tests/test_upload_part_limits.py b/server/tests/test_upload_part_limits.py index 7335685..e971331 100644 --- a/server/tests/test_upload_part_limits.py +++ b/server/tests/test_upload_part_limits.py @@ -42,3 +42,38 @@ def test_an_ordinary_gzip_part_still_lands(): response = client.put(f"/uploads/{upload_id}/parts/0", headers={**AUTH, "Content-Encoding": "gzip"}, content=gzip.compress(b"y" * (1024 * 1024))) assert response.status_code == 200, response.text + + +def test_a_part_is_held_in_memory_once_on_its_way_to_disk(tmp_path, monkeypatch): + """A part was copied whole twice between the socket and the disk -- once + into `bytes`, once by the slice handed to `pwrite` -- so a 64 MB part held + three times that, per part, with parts arriving in parallel.""" + import asyncio + import hashlib + import tracemalloc + + monkeypatch.setattr(settings, "TEMP_DIR", str(tmp_path)) + size = 16 * 1024 * 1024 + session = transfer.create_upload("scan.nii", size, chunk_size=size) + piece = b"x" * 65536 + digest = hashlib.sha256(piece * (size // len(piece))).hexdigest() + + class _Request: + async def stream(self): + for _ in range(size // len(piece)): + yield piece + + async def receive(): + # Measured inside the coroutine, the way the endpoint awaits it: what + # a test harness does with the return value is not the server's. + tracemalloc.start() + try: + body = await main._read_bounded(_Request(), size + 1024 * 1024) + transfer.write_part(session, 0, body, digest) + return tracemalloc.get_traced_memory()[1] + finally: + tracemalloc.stop() + + peak = asyncio.run(receive()) + assert set(session.received_parts()) == {0} + assert peak < 1.3 * size, f"peak {peak / size:.2f}x the part" diff --git a/server/wire/debug_page.py b/server/wire/debug_page.py index 590c664..77efeb6 100644 --- a/server/wire/debug_page.py +++ b/server/wire/debug_page.py @@ -2042,9 +2042,10 @@ } } + // Hands back its fetch, so `tick` can wait for the answer. function load() { - if (!token) { el("gate").hidden = false; el("shell").hidden = true; return; } - fetch("admin-panel.json", { headers: { "X-Admin-Token": token } }) + if (!token) { el("gate").hidden = false; el("shell").hidden = true; return Promise.resolve(); } + return fetch("admin-panel.json", { headers: { "X-Admin-Token": token } }) .then(function (r) { if (r.status === 401) { throw new Error("That is not this server's admin token."); } if (r.status === 403) { throw new Error("This server has no admin token: set ADMIN_TOKEN in its .env and restart it."); } @@ -2058,7 +2059,13 @@ }) .catch(function (e) { el("gate").hidden = false; el("shell").hidden = true; el("gateerr").textContent = e.message; }); } - function tick() { if (live) { load(); } window.setTimeout(tick, EVERY); } + // Re-armed once the poll has SETTLED, never on a clock of its own: against + // a server slow to answer -- exactly when an operator opens this page -- a + // timer re-armed regardless stacked one more panel request every two + // seconds behind those still waiting, and made the server slower still. + function tick() { + (live ? load() : Promise.resolve()).finally(function () { window.setTimeout(tick, EVERY); }); + } el("save").addEventListener("click", function () { token = el("token").value.trim(); @@ -2262,8 +2269,7 @@ readUrl(); drawFilters(); - load(); - window.setTimeout(tick, EVERY); + tick(); })(); diff --git a/server/wire/runs.py b/server/wire/runs.py index ad58cd1..25f5ccb 100644 --- a/server/wire/runs.py +++ b/server/wire/runs.py @@ -48,6 +48,8 @@ import re import secrets import shutil +import sys +import threading import time from typing import List, Optional @@ -1321,6 +1323,132 @@ def progress_file(run_id: Optional[str]) -> Optional[str]: return os.path.abspath(os.path.join(directory, EVENTS_FILE)) +# What a run left in flight by a previous server process is told when the next +# one starts. Said to the client, so it says what happened and nothing else. +ORPHAN_MESSAGE = "The server restarted during this run." + + +def fail_orphans() -> int: + """End every run a previous server process left pending or running. + + TEMP_DIR outlives a restart, and nothing else would ever end such a run: + the process that owned it is gone, so no terminal event is coming, and the + event stream touches the directory every quarter second, so a run whose + client is still watching is never reaped either. That client waits on a + terminal event that cannot arrive, with no timeout of its own. Failing the + run tells it what happened, and the next stream read ends. + + A PAUSED run is left alone: it waits for a person, not for a process, and + a resume picks it up from its job directory whichever process serves it. + + Ending an orphan is not the run being active, so its idle clock is kept as + it was -- the way `extend_pause` keeps it -- and an abandoned one is still + reaped on time. + + **Assumes one server process**, which is what serve.py runs: called from + the lifespan, it would end the live runs of a sibling `uvicorn --workers` + process starting beside them. + """ + ended = 0 + for entry in active(limit=sys.maxsize): + if entry["state"] not in (STATE_PENDING, STATE_RUNNING): + continue + directory = os.path.join(_runs_root(), entry["run_id"]) + try: + idle = os.stat(directory) + except OSError: + continue + finish(entry["run_id"], PHASE_FAILED, message=ORPHAN_MESSAGE) + try: + os.utime(directory, ns=(idle.st_atime_ns, idle.st_mtime_ns)) + except OSError: + pass + ended += 1 + if ended: + logger.info("Ended %d run(s) a previous server process left in flight", ended) + return ended + + +# What the listing knows of each run: directory -> a `_ListedRun` reading that +# run's events forward. `/status`, both admin pages, the update agent's idle +# check and the cap on runs per workstation (every POST /run) all list every +# run, and re-reading a run parsed and re-verified the signature of every +# event it ever wrote: 2.4 ms a run, two and a half seconds a poll with 500 +# runs on disk -- long enough for `/status` to time out, which the update +# agent read as a server that was down. Re-reading only the runs whose file +# had changed was not enough either: a LIVE run changes between any two +# listings, and one with 100k lines cost 634 ms of every /run. +# +# So a run is read on from where the listing stopped, the way the event +# stream tails it. Replaced whole on each listing rather than edited, which +# drops a discarded run's entry with it and lets two concurrent listings each +# publish a complete dict; each entry holds its own lock. +_LISTED: dict = {} + + +class _ListedRun: + """One run as the listing reads it: an `EventReader` kept between + listings, the latest event it delivered, and the nested calls still open. + + The reader is the same one a whole read uses, so what it delivers -- the + signature checked on every new line, the weighted fraction, the cap, the + stop at the terminal event -- is what a whole read would deliver; it is + only given the lines in several goes. The open calls are paired exactly + as `timeline` pairs them, so the chain is `timeline(events)["chain"]`. + """ + + __slots__ = ("lock", "directory", "inode", "identity", "reader", "latest", "open_calls") + + def __init__(self, directory: str, inode: int): + self.lock = threading.Lock() + self.directory = directory + self.inode = inode + self._start() + + def _start(self) -> None: + self.identity = None + # keep_alive=False: LOOKING at a run is not the run being alive. + # Reading stamps the directory, and the TTL is an idle timeout, so an + # operator page polling every couple of seconds would hold every + # abandoned run open for as long as the tab stayed on screen. + self.reader = EventReader(self.directory, keep_alive=False) + self.latest = None + self.open_calls = {} + + def refresh(self, identity: tuple, size: int) -> tuple: + """(latest event, open chain), reading only what was appended.""" + with self.lock: + if identity != self.identity: + if size < self.reader._offset: + # Shorter than what was read: rewritten, not appended to, + # so nothing read before can be trusted. + self._start() + try: + # Past a terminal event a whole read stops, and so does + # this: a tool writing on after it changes nothing. + if not self.reader.finished: + self._fold(self.reader.read()) + self.identity = identity + except Exception: # noqa: BLE001 - a listing must not fail on one bad run + self._start() + return self.latest, self._chain() + + def _fold(self, events: List[dict]) -> None: + for event in events: + if event.get("kind") == LOG_KIND: + continue + self.latest = event + tool = event.get("tool") + if tool: + key = (tool, event.get("depth", 0)) + if self.open_calls.pop(key, None) is None: + self.open_calls[key] = event.get("at") + + def _chain(self) -> list: + still_open = sorted(self.open_calls.items(), key=lambda item: (item[0][1], item[1])) + return [tool for (tool, _depth), _started in still_open] + + def active(limit: int = 500, with_chain: bool = False) -> list: """Every run the registry still holds, newest first, without its events. @@ -1336,11 +1464,13 @@ def active(limit: int = 500, with_chain: bool = False) -> list: on this deployment is every workstation. The phase says what it is doing; the message would say whose data it is doing it to. """ + global _LISTED root = _runs_root() try: names = os.listdir(root) except OSError: return [] + previous, listed = _LISTED, {} found = [] for name in names: if not _scratch.is_valid_id(name): @@ -1349,20 +1479,17 @@ def active(limit: int = 500, with_chain: bool = False) -> list: events_path = os.path.join(directory, EVENTS_FILE) try: started_at = os.path.getctime(directory) - updated_at = os.path.getmtime(events_path) + stat = os.stat(events_path) except OSError: continue - latest = None - events = [] - try: - # keep_alive=False: LOOKING at a run is not the run being alive. - # Reading stamps the directory, and the TTL is an idle timeout, so - # an operator page polling every couple of seconds would hold every - # abandoned run open for as long as the tab stayed on screen. - events = EventReader(directory, keep_alive=False).read() - latest = events[-1] if events else None - except Exception: # noqa: BLE001 - a listing must not fail on one bad run - pass + updated_at = stat.st_mtime + known = previous.get(directory) + if known is None or known.inode != stat.st_ino: + # A new run, or a file replaced under the same name: from the start. + known = _ListedRun(directory, stat.st_ino) + latest, chain = known.refresh((stat.st_ino, stat.st_size, stat.st_mtime_ns), + stat.st_size) + listed[directory] = known tool = None client = None batch = None @@ -1414,7 +1541,10 @@ def active(limit: int = 500, with_chain: bool = False) -> list: # Tool names only, which `_clean_tool_name` already restricted to # identifiers: the operator page may say a run is inside ALI_CBCT # without anything a tool wrote reaching it. - entry["chain"] = timeline(events)["chain"] + entry["chain"] = list(chain) found.append(entry) + # Only directories still listed survive: a discarded or reaped run is + # forgotten on the next poll rather than held for the life of the process. + _LISTED = listed found.sort(key=lambda entry: entry["started_at"], reverse=True) return found[:limit] diff --git a/server/wire/status_page.py b/server/wire/status_page.py index 2f22b73..b21dadf 100644 --- a/server/wire/status_page.py +++ b/server/wire/status_page.py @@ -209,9 +209,13 @@ el("when").textContent = "updated " + new Date().toLocaleTimeString(); } + // Hands back its fetch, so the timer below can wait for the answer. function poll() { - if (!token) { show("gate", true); el("when").textContent = "waiting for a token"; return; } - fetch("status", { headers: { Authorization: "Bearer " + token } }) + if (!token) { + show("gate", true); el("when").textContent = "waiting for a token"; + return Promise.resolve(); + } + return fetch("status", { headers: { Authorization: "Bearer " + token } }) .then(function (r) { if (r.status === 401) { throw new Error("That token was refused."); } if (!r.ok) { throw new Error("The server answered " + r.status + "."); } @@ -235,8 +239,14 @@ if (e.key === "Enter") { el("save").click(); } }); - poll(); - window.setInterval(poll, 2000); + // The next poll is armed once this one has SETTLED, never on a clock of its + // own: against a server slow to answer -- exactly when somebody opens this + // page to see why -- a fixed interval stacked one more request every two + // seconds behind those still waiting, and made it slower still. + function loop() { + poll().finally(function () { window.setTimeout(loop, 2000); }); + } + loop(); }()); diff --git a/server/wire/transfer.py b/server/wire/transfer.py index 1cb9c68..1ee3f2d 100644 --- a/server/wire/transfer.py +++ b/server/wire/transfer.py @@ -263,9 +263,13 @@ def write_part(session: UploadSession, index: int, data: bytes, sha256: Optional file_descriptor = os.open(session.blob_path, os.O_WRONLY) try: + # Through a memoryview: slicing a bytearray copies it, so a short + # write's retry -- and the first write, for `data[0:]` -- would hold + # the part twice. + view = memoryview(data) written = 0 - while written < len(data): - written += os.pwrite(file_descriptor, data[written:], offset + written) + while written < len(view): + written += os.pwrite(file_descriptor, view[written:], offset + written) finally: os.close(file_descriptor) From 62d95a7ddbbc6eb632d30d4393f5611adbb0af44 Mon Sep 17 00:00:00 2001 From: Jules GRIVOT PELISSON Date: Thu, 29 Oct 2026 22:36:59 -0400 Subject: [PATCH 2/2] TEST: read the pages' scripts with a split, not a regex CodeQL takes for an HTML filter --- server/tests/test_page_polling.py | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/server/tests/test_page_polling.py b/server/tests/test_page_polling.py index 0fce3bb..dd631e2 100644 --- a/server/tests/test_page_polling.py +++ b/server/tests/test_page_polling.py @@ -12,7 +12,6 @@ """ import json -import re import shutil import subprocess @@ -99,7 +98,9 @@ def _scripts(page: str) -> str: - return "\n;\n".join(re.findall(r"", page, flags=re.S)) + """The page's own inline scripts, in order. The page is a constant of this + repository, not input: a plain split is all it needs.""" + return "\n;\n".join(chunk.split("", 1)[0] for chunk in page.split("