diff --git a/src/digitaltwin/service/plugin.py b/src/digitaltwin/service/plugin.py index c6fe9a8..f977930 100644 --- a/src/digitaltwin/service/plugin.py +++ b/src/digitaltwin/service/plugin.py @@ -190,6 +190,13 @@ async def register_session(self, request: Request) -> dict: "backends": ["concurrent"]}}} `'learning'` is optional: unconfigured, it aliases `'inference'`. + + A role may name a dispatcher-managed pool instead of an endpoint + (`{"pool": "exsitu"}`), with the pool configs declared once at the + session level (`"pools": [, ...]` -- the task + dispatcher's own schema). Pool-backed roles run on the pool's + pilots, survive single-endpoint loss (tasks requeue; keep them + idempotent), and share one dispatcher session per DT session. """ self._ensure_cleanup_task() diff --git a/src/digitaltwin/service/session.py b/src/digitaltwin/service/session.py index 32a8d66..2e14f05 100644 --- a/src/digitaltwin/service/session.py +++ b/src/digitaltwin/service/session.py @@ -595,13 +595,15 @@ async def _shutdown_engine(self, flow: WorkflowEngine) -> None: self.sid, exc) async def _create_backend(self, name: str): - cfg = self._engine_config(name) + cfg = self._engine_config(name) + pool = cfg.get("pool") log.info( - "[dt] session %s building %r backend on endpoint %s", + "[dt] session %s building %r backend on %s", self.sid, name, - cfg.get("endpoint_name") or "", + f"pool {pool!r}" if pool + else f"endpoint {cfg.get('endpoint_name') or ''}", ) kwargs: dict = dict( @@ -612,22 +614,56 @@ async def _create_backend(self, name: str): name=name, # the asyncflow routing label: task backend= ) + params = inspect.signature(OrbitExecutionBackend.__init__).parameters + # name the backend's broker participant after what it is for, so a # topology view shows `rhapsody..` instead of an # anonymous uuid. Unique by construction: one engine per role per # session (`engine` caches, `_lost` forbids rebuilds). Guarded so # a rhapsody without the parameter keeps working. - if "participant_name" in inspect.signature( - OrbitExecutionBackend.__init__).parameters: + if "participant_name" in params: kwargs["participant_name"] = ( f"rhapsody.{self.sid.split('.')[-1]}.{name}") + if pool: + # A dispatcher-managed pool: the task dispatcher runs on the + # broker, and the pool's pilots pick the executing endpoints -- + # this backend targets no endpoint of its own. One dispatcher + # session per DT session, keyed by this session's own sid and + # shared by every role that names a pool; the session-level + # `pools` configs are declared with it, and re-declaring an + # owned pool is idempotent, so role build order does not + # matter. + # + # Contract: a task lost with its pilot is requeued and + # re-executes, so twin tasks must stay idempotent. In return + # the twin no longer dies with a single endpoint -- pool-backed + # roles opt out of the R8 fail-fast (see `endpoints_lost`), and + # a task that ultimately fails surfaces through the normal + # task-status path. + if "pool" not in params: + raise RuntimeError( + f"engine {name!r} names pool {pool!r}, but the installed" + " rhapsody's OrbitExecutionBackend has no pool support") + session_kwargs: dict = {"sid": self.sid} + if self.config.get("pools"): + session_kwargs["pools"] = self.config["pools"] + kwargs.update( + plugin_name="task_dispatcher", + pool=pool, + session_kwargs=session_kwargs, + ) + backend = await OrbitExecutionBackend(**kwargs) - # the endpoint the backend *settled on* (it auto-selects when the - # config named none) -- what a topology change is matched against + # What a topology change is matched against: the endpoint the + # backend *settled on* (it auto-selects when the config named + # none). A pool-backed role records its pool instead -- no lost + # endpoint ever matches it, which is exactly the R8 opt-out. self._endpoints[name] = ( - getattr(backend, "_endpoint_name", None) or cfg.get("endpoint_name") + f"pool:{pool}" if pool + else getattr(backend, "_endpoint_name", None) + or cfg.get("endpoint_name") ) return backend @@ -645,6 +681,12 @@ def endpoints_lost(self, lost: set[str]) -> tuple[str, ...]: The loss is also remembered, because the broker announces it only once -- see `engine`. + + A pool-backed role never matches: its `_endpoints` entry is + `pool:`, not an endpoint. Losing a pilot's endpoint there + requeues the pilot's tasks inside the dispatcher instead of + killing the twin -- the twin's failure surface for pools is a task + that ultimately fails, not a single endpoint going away. """ names = set(self._endpoints) | set(self.config.get("engines") or {}) diff --git a/test/unit/test_service.py b/test/unit/test_service.py index 6dd19fe..42e735b 100644 --- a/test/unit/test_service.py +++ b/test/unit/test_service.py @@ -817,3 +817,88 @@ def test_the_explorer_module_is_the_one_the_plugin_declares(): assert module.is_file() assert module.name in UI_ASSETS assert "window.DTDash.mount" in module.read_text() + + +# --------------------------------------------------------------------------- +# pool-backed roles (dispatcher pools) +# --------------------------------------------------------------------------- + +class _FakePoolBackend: + """Signature-faithful stand-in: the pool path checks the constructor's + parameters before using them.""" + + captured: dict = {} + + def __init__(self, broker_url=None, endpoint_name=None, backends=None, + name=None, participant_name=None, plugin_name="rhapsody", + pool=None, session_kwargs=None, batch_window=0): + type(self).captured = dict( + endpoint_name=endpoint_name, name=name, + participant_name=participant_name, plugin_name=plugin_name, + pool=pool, session_kwargs=session_kwargs) + self._endpoint_name = endpoint_name or "broker" + + def __await__(self): + async def _self(): + return self + return _self().__await__() + + +class _FakeOldBackend: + """A backend from before pool support -- no `pool` parameter.""" + + def __init__(self, broker_url=None, endpoint_name=None, backends=None, + name=None, batch_window=0): + pass + + +async def test_a_pool_backed_role_targets_the_dispatcher(monkeypatch): + """`pool` in a role's config routes it at the task dispatcher: one + dispatcher session per DT session (keyed by the DT sid), the + session-level pool configs declared with it.""" + + import digitaltwin.service.session as session_mod + + session = DTSession("s1") + session.config = { + "pools": [{"name": "exsitu", "endpoint_name": "hpc1"}], + "engines": {"inference": {"endpoint_name": "ep1"}, + "learning": {"pool": "exsitu"}}, + } + monkeypatch.setattr(session_mod, "OrbitExecutionBackend", + _FakePoolBackend) + + await session._create_backend("learning") + + got = _FakePoolBackend.captured + assert got["plugin_name"] == "task_dispatcher" + assert got["pool"] == "exsitu" + assert got["session_kwargs"]["sid"] == "s1" + assert got["session_kwargs"]["pools"] == [ + {"name": "exsitu", "endpoint_name": "hpc1"}] + # the role records its pool, not an endpoint -- the R8 opt-out + assert session._endpoints["learning"] == "pool:exsitu" + + +async def test_a_pool_needs_a_pool_capable_backend(monkeypatch): + import digitaltwin.service.session as session_mod + + session = DTSession("s1") + session.config = {"engines": {"learning": {"pool": "exsitu"}}} + monkeypatch.setattr(session_mod, "OrbitExecutionBackend", + _FakeOldBackend) + + with pytest.raises(RuntimeError, match="no pool support"): + await session._create_backend("learning") + + +async def test_endpoint_loss_spares_pool_backed_roles(): + """Losing an endpoint requeues a pool's tasks inside the dispatcher; + only endpoint-bound roles fail their twins (R8).""" + + session = DTSession("s1") + session.config = {"engines": {"learning": {"pool": "exsitu"}}} + session._endpoints["learning"] = "pool:exsitu" + + assert session.endpoints_lost({"hpc1", "exsitu", "pool"}) == () + assert not session._lost