diff --git a/src/dls_tiled/visr.py b/src/dls_tiled/visr.py index 0976cb0..e84f11e 100644 --- a/src/dls_tiled/visr.py +++ b/src/dls_tiled/visr.py @@ -104,6 +104,27 @@ async def _shared(key: tuple, compute): return await asyncio.shield(task) +async def _run_progress(root, uid) -> tuple: + """How far a run has got, as its catalog records it: whether it has a stop + document, and the shape of its camera's ``RedTotal`` (one entry per point). + + Part of the key of a shared result, so a result is never reused once the run + has moved on. Without it, the poll a UI makes when a scan ends could be + given the result of the poll just before, computed without the last points. + Anything that can't be looked up counts as unknown (``None``).""" + try: + run = await root.lookup_adapter([uid]) + stopped = bool((run.metadata() or {}).get("stop")) + except Exception: + return (None, None) + try: + camera = await root.lookup_adapter([uid, "primary", "RedTotal"]) + shape = tuple(camera.structure().shape) + except Exception: + shape = None + return (stopped, shape) + + async def _compute_and_keep(key: tuple, compute): try: result = await compute() @@ -653,12 +674,17 @@ async def binned( # type: ignore # A UI polls this several times a second per client while points arrive only # every couple of seconds, so identical requests share one computation (and - # one read of the files) and its result is reused for a moment. + # one read of the files) and its result is reused for a moment, but only + # while the run is where it was (see _run_progress). # The key leaves out who is asking. That is safe only because the data is # read through root_tree.lookup_adapter, which bypasses Tiled's access policy # anyway; if per-run access control is added, the key needs the principal # (or its access tags), or one user's result could be served to another. - key = (uid, tuple(sorted(request.query_params.multi_items()))) + key = ( + uid, + tuple(sorted(request.query_params.multi_items())), + await _run_progress(root, uid), + ) async def compute(): output = await _binned_output( @@ -1167,7 +1193,12 @@ async def render( # type: ignore root = request.app.state.root_tree uid = [s for s in path.strip("/").split("/") if s][0] # Shared like /binned's results, and with the same caveat about access control. - key = ("render", uid, tuple(sorted(request.query_params.multi_items()))) + key = ( + "render", + uid, + tuple(sorted(request.query_params.multi_items())), + await _run_progress(root, uid), + ) async def compute(): points = await load_points( diff --git a/tests/test_visr.py b/tests/test_visr.py index f26ef45..bb99041 100644 --- a/tests/test_visr.py +++ b/tests/test_visr.py @@ -636,6 +636,63 @@ async def scenario(): assert compute.calls == 2 +def grow_grid_scan(client, uid, n_from, n_to): + """Append points n_from..n_to of write_grid_scan's raster to a run, as a + scan in progress does""" + primary = client[uid]["primary"] + x = numpy.array([0.0, 1.0, 2.0] * 3)[n_from:n_to] + y = numpy.repeat([0.0, 1.0, 2.0], 3)[n_from:n_to] + primary["X"].patch(x, offset=n_from, extend=True) + primary["Y"].patch(y, offset=n_from, extend=True) + for channel in ("RedTotal", "GreenTotal", "BlueTotal"): + primary[channel].patch( + numpy.arange(9.0)[n_from:n_to], offset=n_from, extend=True + ) + + +@pytest.mark.parametrize("route", ["binned", "render"]) +def test_a_shared_result_is_not_reused_once_the_run_has_grown( + client, monkeypatch, route +): + monkeypatch.setattr(visr, "BINNED_CACHE_SECONDS", 60.0) + write_grid_scan(client, "growing", 7) + http = client.context.http_client + + before = http.get(f"/api/v1/{route}/growing").json() + grow_grid_scan(client, "growing", 7, 9) + after = http.get(f"/api/v1/{route}/growing").json() + + assert (before["n_points"], after["n_points"]) == (7, 9) + + +def test_a_shared_result_is_not_reused_once_the_run_has_stopped(client, monkeypatch): + monkeypatch.setattr(visr, "BINNED_CACHE_SECONDS", 60.0) + write_grid_scan(client, "ending", 9) + http = client.context.http_client + compute = [] + real = visr._binned_output + + async def counting(*args): + compute.append(1) + return await real(*args) + + monkeypatch.setattr(visr, "_binned_output", counting) + + http.get("/api/v1/binned/ending") + http.get("/api/v1/binned/ending") # reused + run = client["ending"] + run.update_metadata(metadata={**run.metadata, "stop": {"exit_status": "success"}}) + http.get("/api/v1/binned/ending") + + assert len(compute) == 2 + + +def test_run_progress_is_unknown_for_a_run_that_is_not_there(client): + root = client.context.http_client.app.state.root_tree + + assert asyncio.run(visr._run_progress(root, "nowhere")) == (None, None) + + def test_binned_has_an_etag_and_answers_304_when_nothing_changed(client, monkeypatch): monkeypatch.setattr(visr, "BINNED_CACHE_SECONDS", 0.5) write_grid_scan(client, "grid", 9)