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..dd631e2 --- /dev/null +++ b/server/tests/test_page_polling.py @@ -0,0 +1,138 @@ +"""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 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: + """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(" 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)