From d06e116c15102e3ff56e592dd0cd241858e2acad Mon Sep 17 00:00:00 2001 From: ykcheungab Date: Mon, 14 Sep 2026 20:53:38 +0800 Subject: [PATCH] =?UTF-8?q?fix(studio-3d):=20=E5=8F=96=E6=B6=88=20epoch=20?= =?UTF-8?q?=E2=80=94=E2=80=94=20=E5=81=9C=E6=AD=A2/=E9=9A=90=E8=97=8F/?= =?UTF-8?q?=E6=8D=A2=E4=BC=9A=E8=AF=9D=E4=B9=8B=E5=90=8E=EF=BC=8C=E5=B7=B2?= =?UTF-8?q?=E5=9C=A8=E8=B7=91=E7=9A=84=E5=90=8E=E5=8F=B0=E5=90=88=E6=88=90?= =?UTF-8?q?=E4=B8=8D=E8=83=BD=E5=86=8D=E6=8A=8A=E6=97=A7=E6=B5=8B=E9=87=8F?= =?UTF-8?q?=E6=94=BE=E5=9B=9E=E5=B1=8F=E5=B9=95?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - What: 1. `ThreeDCoordinator` 新增 **epoch**（显示声明代次）与 `_Inflight` 记录：`work()` 不再直接写 `_composed` mailbox，只 `return` 合成好的帧；结果在 GUI 线程的 `_collect_inflight()` 里被接受， 且必须同时满足 `item.epoch == self._epoch` 与 `_identity_is_current(identity)` 才会入队。 `activate()`/`begin_session()`/`set_visible(False)`/`show_rest()`/`close()` 统一走新的 `_invalidate()`：epoch+1、清空两块 mailbox、清 `_last_request_id`/`_last_composed_identity`/ `_last_field_frame`/`_settings_dirty`，并**尝试**取消 future。 2. **未成功取消的 future 仍占用唯一合成槽**：`_inflight` 只在 `future.done()` 时由 `_collect_inflight()` 释放，所以反复隐藏/显示不会在线程池里排起第二个合成；`CancelledError` 单独吞掉（取消不是失败， 不产生诊断），其它异常仍记成 `field_unavailable` 诊断（文案由 "compose raised" 回到 "compose failed"）。 3. **request_id 单调**：`submit()` 拒绝小于本 epoch 已见值的 request_id（>0 时）并计 `stale` + 给 `stale_frame` 诊断；`_identity_is_current()` 同样比较，所以"晚到的旧结果"也进不了屏幕。 4. 新增 `ThreeDCoordinator.epoch` 与 `last_presented()`（帧号 + 单调时刻，供后续面板帧计数用）； 构造参数新增关键字 `clock`（默认 `time.monotonic`）与 `stale_grace_s`（默认 1.0，本提交尚未使用）。 - Why: 审查报告 P1/R1 的第一条。`Future.cancel()` 无法中断已经开始运行的函数，而旧代码在 `_cancel_inflight()` 里就把 `_inflight` 置空并只比较 generation —— 于是"停止显示"之后旧 work 结束仍然 `offer` 进 `_composed`，下一次 tick 取到它并把状态改回 live（审查用可控后台任务复现： `valid_after_show_rest=false` → 放行旧 compose → `valid_after_old_worker_finishes=true`、`state=live`）。 同一个洞也适用于"停止重算后定时器仍继续 tick"的场景。把接受点移到 GUI 线程后，"发布前校验"和 "接收时校验"是同一处代码，epoch 让"这一刻的显示声明"成为可比较的事实。 - Verification: · `test_scene3d_coordinator.py` +7（24 通过）：合成已进入后被 `show_rest`/`set_visible(False)`/ `begin_session` 撤销 → 旧结果永不 present 且 state 不为 live、epoch 递增、槽位在 work 真正结束后释放； `ManualExecutor` 复现"提交后立即取消" → `CancelledError` 不产生诊断且槽位可复用； 20 轮 begin_session→submit→tick→show_rest 循环里最后一个画面上永远没有有效测量、`max_in_flight==1`； request_id 逆序在入口被拒（`stale` + `stale_frame`）且画面停在较新的请求上， "较旧请求的结果晚到"也在接收侧被丢弃。 · 审查复现脚本对照（`.repro/probe_stage6_r1.py` → `probe_stage6_r1.json`）： `stopped_frame_returns` 由 `{valid_after_old_worker_finishes: true, state: "live"}` 变为 `{false, "no_data", epoch_bumped: true, slot_free_again: true}`； `request_id_not_checked` 由 `{old_submit: "accepted", shown_request: 1}` 变为 `{old_submit: "stale", shown_request: 2, diagnostic: "stale_frame"}`。 · apps/studio 全量 pytest 退出码 0（含真 GL 冒烟用例）。 - Not in this change: `stale_grace_s`/`clock` 只落地参数与 `last_presented()` 的记录，**失效状态的迁移 （`stale` 状态、宽限到期回参考模型）留给下一提交**；adapter 侧的字段级有效性、设备几何身份、 资产加载请求号、二维后端统一均未动；审查报告里那句"用 `coordinator._inflight` 当 Future 用"的复现 脚本无法再原样运行（槽位现在是 `_Inflight` 记录），改用上面那份等价探针。 --- .../src/orisys_studio/scene3d/coordinator.py | 186 ++++++++++---- apps/studio/tests/test_scene3d_coordinator.py | 227 +++++++++++++++++- 2 files changed, 359 insertions(+), 54 deletions(-) diff --git a/apps/studio/src/orisys_studio/scene3d/coordinator.py b/apps/studio/src/orisys_studio/scene3d/coordinator.py index 4847e83..6e77c24 100644 --- a/apps/studio/src/orisys_studio/scene3d/coordinator.py +++ b/apps/studio/src/orisys_studio/scene3d/coordinator.py @@ -20,13 +20,25 @@ Staleness is about *identity*, not timing: a frame from another generation, anot session or another source than the active scene is dropped with ``stale_frame`` and counted, and an old result can never overwrite a new scene because the composer's result carries the identity it was built from and is checked again on arrival. + +**A withdrawn display claim is an epoch, not a coincidence.** ``Future.cancel()`` +cannot stop a function that has already started, so "stop", "hide", "new session" and +"new scene" each bump an epoch that the running compose captured when it started. The +work itself only *returns* its frame; the GUI thread accepts it, and only if the epoch +and the identity it carries are still the current ones. Without that, a compose that +was already running when the operator pressed stop finished a moment later and put the +old measurement back on screen while the view said "live" again — the exact bug the +epoch exists to prevent. The slot stays occupied until the work really finishes, so +repeated hiding and showing cannot queue up composes behind a slow one. """ from __future__ import annotations import threading -from concurrent.futures import Executor, Future -from typing import Any, Optional +import time +from concurrent.futures import CancelledError, Executor, Future +from dataclasses import dataclass +from typing import Any, Callable, Optional from orisys_app_shared.scene3d.contracts import ( Diagnostic, @@ -46,6 +58,25 @@ from orisys_app_shared.scene3d.contracts import ( #: counter is informational, never a reason to throttle acquisition. MAX_INFLIGHT_COMPOSES = 1 +#: How long a measurement may fail to arrive before the display stops claiming it is +#: live. One dropped frame is not a reason to blank the picture; a stream of them is. +DEFAULT_STALE_GRACE_S = 1.0 + + +@dataclass(frozen=True) +class _Inflight: + """One compose task, and the display claim it was started under. + + Kept as a whole rather than as a bare future so that "which claim does this + result belong to?" can be asked *before* the result is published, and so that a + future which refused to cancel still occupies the single compose slot. + """ + + future: Future + epoch: int + identity: FrameIdentity + started: float + class ThreeDCoordinator: """Owns the two mailboxes, the compose worker and the scene generation.""" @@ -56,28 +87,42 @@ class ThreeDCoordinator: composer: Any, executor: Executor, max_fps: int = 30, + *, + clock: Callable[[], float] = time.monotonic, + stale_grace_s: float = DEFAULT_STALE_GRACE_S, ) -> None: self._widget = widget self._composer = composer self._executor = executor + self._clock = clock + self._stale_grace_s = max(0.0, float(stale_grace_s)) self._max_fps = max(1, int(max_fps)) self._incoming: LatestFrameMailbox[SurfaceFieldFrame] = LatestFrameMailbox() self._composed: LatestFrameMailbox[RenderFrame] = LatestFrameMailbox() self._lock = threading.Lock() self._state: SceneState = "disabled" self._generation = 0 + #: Which display claim the frames belong to. Bumped whenever the claim is + #: withdrawn: a stop, a hide, a new session, a new scene. A compose result + #: whose epoch is not this one is thrown away instead of presented. + self._epoch = 0 self._assets: Optional[SceneAssets] = None self._source_id = "" self._session_id: Optional[int] = None + #: The newest request id seen in this epoch: a producer that falls behind + #: (a seek UI, a queue that reorders) must not put an older frame on screen. + self._last_request_id = 0 self._visible = False self._settings = DisplaySettings(max_fps=self._max_fps) - self._inflight: Optional[Future] = None + self._inflight: Optional[_Inflight] = None self._stale_frames = 0 self._invalid_frames = 0 self._last_diagnostic: Optional[Diagnostic] = None self._closed = False self._settings_dirty = False self._last_composed_identity: Optional[FrameIdentity] = None + #: What is on screen and when it got there, for the pane's frame counter. + self._last_presented: Optional[tuple[FrameIdentity, float]] = None #: The field frame currently on screen, so a gain change can redraw exactly #: it rather than a newer instant. self._last_field_frame: Optional[SurfaceFieldFrame] = None @@ -92,6 +137,15 @@ class ThreeDCoordinator: def generation(self) -> int: return self._generation + @property + def epoch(self) -> int: + """The display claim results are accepted under (see :class:`_Inflight`).""" + return self._epoch + + def last_presented(self) -> Optional[tuple[FrameIdentity, float]]: + """What is on screen, and the monotonic time it was handed to the widget.""" + return self._last_presented + @property def superseded_frames(self) -> int: """Frames the display never showed. Never a recording drop.""" @@ -121,6 +175,28 @@ class ThreeDCoordinator: self._state = state # --------------------------------------------------------------- lifetime + def _invalidate(self) -> None: + """Withdraw the current display claim: nothing in flight may change the view. + + Called by every transition that means "what was on screen is no longer being + claimed": stop, hide, a new session, a new scene, close. The epoch is what + makes a compose that is *already running* harmless — it will finish, its + result will be recognised as belonging to a withdrawn claim, and it will be + dropped. The single compose slot is deliberately **not** freed here: a future + that refused to cancel still owns it until it really finishes, so repeatedly + hiding and showing cannot build a queue behind a slow composer. + """ + self._epoch += 1 + self._incoming.clear() + self._composed.clear() + self._last_request_id = 0 + self._last_composed_identity = None + self._last_field_frame = None + self._settings_dirty = False + item = self._inflight + if item is not None: + item.future.cancel() + def activate(self, assets: SceneAssets, source_id: str) -> int: """Adopt a scene and a data source; returns the new generation. @@ -129,16 +205,11 @@ class ThreeDCoordinator: """ if self._closed: raise error("asset_rejected", "this coordinator has been closed") - self._cancel_inflight() - self._incoming.clear() - self._composed.clear() + self._invalidate() self._assets = assets self._source_id = str(source_id) self._session_id = None self._generation += 1 - self._last_composed_identity = None - self._last_field_frame = None - self._settings_dirty = False self._widget.set_scene(assets) if self._visible: self._widget.present(self._composer.rest_frame(assets, self._identity(0))) @@ -155,9 +226,7 @@ class ThreeDCoordinator: self._widget.setVisible(visible) if not visible: self._visible = False - self._incoming.clear() - self._composed.clear() - self._cancel_inflight() + self._invalidate() self._set_state("suspended") return self._visible = True @@ -179,14 +248,9 @@ class ThreeDCoordinator: """ if self._closed: return - self._cancel_inflight() - self._incoming.clear() - self._composed.clear() + self._invalidate() self._session_id = None self._generation += 1 - self._last_composed_identity = None - self._last_field_frame = None - self._settings_dirty = False if self._assets is not None: # The reference model, not the last measurement: the new session has not # produced anything yet, and an old picture would be a claim about it. @@ -226,6 +290,19 @@ class ThreeDCoordinator: scene=str(session_id), ) return "stale" + newest_request = self._last_request_id + if identity.request_id > 0 and identity.request_id < newest_request: + # Request ids are monotonic within a session, so an older one is a frame + # the producer has already superseded (a seek, a reordered queue). Showing + # it would move the picture backwards in time. + self._stale_frames += 1 + self._last_diagnostic = diagnostic( + "stale_frame", + "a frame with an older request id was dropped", + frame=str(identity.request_id), + scene=str(newest_request), + ) + return "stale" problem = self._reject_reason(frame, assets) if problem is not None: self._invalid_frames += 1 @@ -234,6 +311,8 @@ class ThreeDCoordinator: with self._lock: if session_id is None: self._session_id = identity.session_id + if identity.request_id > 0: + self._last_request_id = max(self._last_request_id, identity.request_id) return self._incoming.offer(frame) @staticmethod @@ -294,6 +373,7 @@ class ThreeDCoordinator: self._widget.set_superseded_frames(self.superseded_frames) self._widget.present(composed) self._last_composed_identity = composed.identity + self._last_presented = (composed.identity, self._clock()) self._set_state("live" if composed.valid.any() else "no_data") elif self._state in ("ready", "loading"): # Nothing has arrived yet: show the reference model rather than a blank @@ -327,10 +407,7 @@ class ThreeDCoordinator: """ if self._closed or self._assets is None or not self._visible: return - self._incoming.clear() - self._cancel_inflight() - self._last_field_frame = None - self._last_composed_identity = None + self._invalidate() self._widget.present(self._composer.rest_frame(self._assets, self._identity(0))) self._set_state("no_data") @@ -339,9 +416,7 @@ class ThreeDCoordinator: if self._closed: return self._closed = True - self._cancel_inflight() - self._incoming.clear() - self._composed.clear() + self._invalidate() self._set_state("disabled") # ------------------------------------------------------------------ internal @@ -360,6 +435,10 @@ class ThreeDCoordinator: identity.source_id == self._source_id and identity.generation == self._generation and (self._session_id is None or identity.session_id == self._session_id) + and not ( + identity.request_id > 0 + and identity.request_id < self._last_request_id + ) ) def _start_compose(self, frame: SurfaceFieldFrame) -> None: @@ -367,44 +446,45 @@ class ThreeDCoordinator: settings = self._settings if assets is None: return - generation = self._generation + epoch = self._epoch # Kept so a display-settings change can recompose *this* frame instead of # jumping to a newer one. self._last_field_frame = frame - def work() -> None: - try: - composed = self._composer.compose(assets, frame, settings) - except Exception as exc: # noqa: BLE001 - a failed compose is not a crash - self._last_diagnostic = getattr(exc, "diagnostic", None) or diagnostic( - "field_unavailable", f"compose failed: {type(exc).__name__}: {exc}" - ) - return - if composed.identity.generation != generation: - # The scene changed while this was running: the result is for a - # scene that no longer exists. - self._stale_frames += 1 - return - self._composed.offer(composed) + def work() -> RenderFrame: + # No mailbox write here: the result is accepted on the GUI thread, where + # the current epoch can be compared without racing the operator's stop. + return self._composer.compose(assets, frame, settings) - self._inflight = self._executor.submit(work) + self._inflight = _Inflight( + future=self._executor.submit(work), + epoch=epoch, + identity=frame.identity, + started=self._clock(), + ) def _collect_inflight(self) -> None: - future = self._inflight - if future is None: - return - if not future.done(): + item = self._inflight + if item is None or not item.future.done(): return self._inflight = None try: - future.result() + composed = item.future.result() + except CancelledError: + # The work never started, so there is nothing to report: the epoch bump + # that cancelled it is already the reason the display moved on. + return except Exception as exc: # noqa: BLE001 self._last_diagnostic = getattr(exc, "diagnostic", None) or diagnostic( - "field_unavailable", f"compose raised: {type(exc).__name__}: {exc}" + "field_unavailable", f"compose failed: {type(exc).__name__}: {exc}" ) - - def _cancel_inflight(self) -> None: - future = self._inflight - self._inflight = None - if future is not None: - future.cancel() + return + if composed is None: + return + if item.epoch != self._epoch or not self._identity_is_current(composed.identity): + # A result from a claim that has been withdrawn (stop, hide, new session, + # new scene, older request). Presenting it would put a measurement back on + # screen for something that is no longer being measured. + self._stale_frames += 1 + return + self._composed.offer(composed) diff --git a/apps/studio/tests/test_scene3d_coordinator.py b/apps/studio/tests/test_scene3d_coordinator.py index 47b61bf..4680dfd 100644 --- a/apps/studio/tests/test_scene3d_coordinator.py +++ b/apps/studio/tests/test_scene3d_coordinator.py @@ -10,7 +10,8 @@ from __future__ import annotations import threading import time -from concurrent.futures import ThreadPoolExecutor +from concurrent.futures import Future, ThreadPoolExecutor +from dataclasses import replace import numpy as np import pytest @@ -415,3 +416,227 @@ def test_a_frame_already_queued_is_dropped_when_the_model_comes_back(scene, pool "a frame that was in flight when the pass stopped was presented anyway" ) coordinator.close() + + +# ------------------------------------------- a withdrawn claim, and the work behind it +class BlockingComposer(SceneComposer): + """A composer that has *entered* compose and waits for permission to finish. + + This is the shape the audit reproduced: the work is already running when the + operator stops the display, so ``Future.cancel()`` cannot stop it and only the + epoch can make its result harmless. + """ + + def __init__(self) -> None: + super().__init__() + self.entered = threading.Event() + self.release = threading.Event() + + def compose(self, assets, frame, settings): + self.entered.set() + assert self.release.wait(5.0), "the test never released the blocking composer" + return super().compose(assets, frame, settings) + + +class ImmediateExecutor: + """Runs the work inline: makes the schedule deterministic without threads.""" + + def submit(self, function): + future: Future = Future() + try: + future.set_result(function()) + except BaseException as exc: # noqa: BLE001 - mirrored into the future + future.set_exception(exc) + return future + + +class ManualExecutor: + """Futures that only run when the test says so (and honour a cancel first).""" + + def __init__(self) -> None: + self.pending: list[tuple[Future, object]] = [] + + def submit(self, function): + future: Future = Future() + self.pending.append((future, function)) + return future + + def run_next(self) -> None: + """Run the oldest task that was not cancelled, like a real worker would.""" + while self.pending: + future, function = self.pending.pop(0) + if future.cancelled(): + continue + if future.set_running_or_notify_cancel(): + try: + future.set_result(function()) + except BaseException as exc: # noqa: BLE001 - mirrored into the future + future.set_exception(exc) + return + + +def _start_blocked_compose(scene, pool, source="replay-A", value=5.0): + """Set up a coordinator whose first compose is running and blocked.""" + composer = BlockingComposer() + widget = FakeWidget() + coordinator = ThreeDCoordinator(widget, composer, pool) + coordinator.set_visible(True) + coordinator.activate(scene, source) + coordinator.submit(_frame(scene, source=source, value=value)) + coordinator.tick() + assert composer.entered.wait(3.0), "the compose never started" + assert coordinator._inflight is not None + return composer, widget, coordinator + + +def test_a_compose_running_when_the_pass_stopped_can_never_come_back(scene, pool): + """R1: "stopped, then the old worker finished and the pane said live again".""" + composer, widget, coordinator = _start_blocked_compose(scene, pool) + epoch = coordinator.epoch + + coordinator.show_rest() + assert coordinator.state == "no_data" + assert not widget.frames[-1].valid.any(), "stop did not clear the measurement" + + composer.release.set() + _drain(coordinator, ticks=40, sleep=0.01) + + assert coordinator.epoch == epoch + 1 + assert coordinator._inflight is None, "the finished work still owns the slot" + assert not widget.frames[-1].valid.any(), ( + "the stopped pass put its measurement back on screen" + ) + assert coordinator.state != "live" + coordinator.close() + + +def test_hiding_while_a_compose_runs_does_not_let_it_paint_later(scene, pool): + composer, widget, coordinator = _start_blocked_compose(scene, pool) + coordinator.set_visible(False) + assert coordinator.state == "suspended" + composer.release.set() + # Hidden: the tick that would present the result does not even run. + for _ in range(10): + coordinator.tick() + time.sleep(0.005) + coordinator.set_visible(True) + _drain(coordinator, ticks=40, sleep=0.01) + assert not widget.frames[-1].valid.any(), ( + "a compose started before the view was hidden painted after it came back" + ) + assert coordinator.state != "live" + coordinator.close() + + +def test_a_compose_running_for_the_old_session_is_dropped_by_begin_session(scene, pool): + composer, widget, coordinator = _start_blocked_compose(scene, pool) + coordinator.begin_session() + assert widget.frames[-1].scalar_descriptor.quantity == "unavailable" + composer.release.set() + _drain(coordinator, ticks=40, sleep=0.01) + assert not widget.frames[-1].valid.any(), ( + "the previous session's compose was presented into the new session" + ) + coordinator.close() + + +def test_a_compose_cancelled_before_it_started_is_not_an_error(scene): + """A future that never ran is not a failure, and it does not hold the slot.""" + executor = ManualExecutor() + widget = FakeWidget() + coordinator = ThreeDCoordinator(widget, SceneComposer(), executor) + coordinator.set_visible(True) + coordinator.activate(scene, "live-A") + coordinator.submit(_frame(scene, value=4.0)) + coordinator.tick() + assert len(executor.pending) == 1 + + coordinator.show_rest() + assert executor.pending[0][0].cancelled() + coordinator.tick() + assert coordinator.last_diagnostic is None, "a cancelled compose was reported as a failure" + assert coordinator._inflight is None, "the cancelled work still owns the slot" + + coordinator.submit(_frame(scene, frame_id=2, value=4.0)) + coordinator.tick() + executor.run_next() + coordinator.tick() + assert coordinator.state == "live" + assert np.allclose(np.asarray(widget.frames[-1].scalar), 4.0) + coordinator.close() + + +def test_repeated_start_and_stop_never_leaves_an_old_measurement(scene, pool): + """Twenty stop/start cycles: one compose at a time, and never a reanimated one.""" + composer = CountingComposer(delay_s=0.005) + widget = FakeWidget() + coordinator = ThreeDCoordinator(widget, composer, pool) + coordinator.set_visible(True) + coordinator.activate(scene, "live-A") + for index in range(20): + coordinator.begin_session() + coordinator.submit( + _frame( + scene, + frame_id=index, + value=float(index + 1), + generation=coordinator.generation, + ) + ) + coordinator.tick() + coordinator.show_rest() + assert not widget.frames[-1].valid.any(), f"cycle {index} kept a measurement" + _drain(coordinator, ticks=60, sleep=0.01) + assert not widget.frames[-1].valid.any(), "a stopped pass painted after the fact" + assert coordinator.state != "live" + assert composer.max_in_flight == 1 + coordinator.close() + + +def test_an_older_request_id_cannot_move_the_picture_backwards(scene): + """R7: request ids are monotonic within a session, and the pane must obey.""" + widget = FakeWidget() + coordinator = ThreeDCoordinator(widget, SceneComposer(), ImmediateExecutor()) + coordinator.set_visible(True) + coordinator.activate(scene, "live-A") + newer = _frame(scene, frame_id=20, value=2.0) + newer = replace(newer, identity=replace(newer.identity, request_id=2)) + older = _frame(scene, frame_id=10, value=9.0) + older = replace(older, identity=replace(older.identity, request_id=1)) + + assert coordinator.submit(newer) == "accepted" + coordinator.tick() + coordinator.tick() + assert widget.frames[-1].identity.request_id == 2 + + assert coordinator.submit(older) == "stale" + assert coordinator.last_diagnostic is not None + assert coordinator.last_diagnostic.code == "stale_frame" + coordinator.tick() + coordinator.tick() + assert widget.frames[-1].identity.request_id == 2, "an older frame was shown" + assert np.allclose(np.asarray(widget.frames[-1].scalar), 2.0) + assert coordinator.stale_frames == 1 + coordinator.close() + + +def test_a_result_from_an_older_request_is_dropped_on_arrival(scene): + """The same rule on the way out: the composer's result carries the request id.""" + executor = ManualExecutor() + widget = FakeWidget() + coordinator = ThreeDCoordinator(widget, SceneComposer(), executor) + coordinator.set_visible(True) + coordinator.activate(scene, "live-A") + old = _frame(scene, frame_id=10, value=7.0) + old = replace(old, identity=replace(old.identity, request_id=1)) + coordinator.submit(old) + coordinator.tick() # starts the compose for request 1, and it stays pending + newer = _frame(scene, frame_id=20, value=3.0) + newer = replace(newer, identity=replace(newer.identity, request_id=2)) + assert coordinator.submit(newer) == "accepted" + executor.run_next() # request 1's compose finishes late + _drain(coordinator, ticks=6, sleep=0.0) + assert not np.allclose(np.asarray(widget.frames[-1].scalar), 7.0), ( + "a result for a superseded request was presented" + ) + coordinator.close()