diff --git a/src/digitaltwin/runtime.py b/src/digitaltwin/runtime.py index f40a8cc..96f8ea4 100644 --- a/src/digitaltwin/runtime.py +++ b/src/digitaltwin/runtime.py @@ -1053,7 +1053,8 @@ def _ensure_dtype_queue(self, dtype: DataType) -> asyncio.Queue: return self.dtype_queues[dtype] - def add_input(self, dtype: DataType, channel: str, codec: str = CODEC_JSON): + def add_input(self, dtype: DataType, channel: str, + codec: str = CODEC_JSON) -> Optional[asyncio.Task]: """Open the graph at its input edge: bind an external channel. Sensors and other producers live outside the framework. They @@ -1070,6 +1071,10 @@ def add_input(self, dtype: DataType, channel: str, codec: str = CODEC_JSON): Internal producers keep their own path: a persistent component publishes through `RuntimeAPI.stream`. + + Returns the subscription task (`None` for an idempotent re-bind, + or when the twin is tearing down): a caller that needs the + binding live before producers publish awaits it. """ self._check_mutable() @@ -1077,15 +1082,24 @@ def add_input(self, dtype: DataType, channel: str, codec: str = CODEC_JSON): PubSubClient.check_channel(channel) check_codec(codec) - binding = _InputBinding(dtype, channel, codec) - if binding in self.inputs: - return - self.inputs.append(binding) + for existing in self.inputs: + if (existing.dtype, existing.channel) == (dtype, channel): + if existing.codec == codec: + return None + # the stream client dedupes on (channel, dtype), so a + # changed codec would be recorded here yet never applied + # -- refuse it rather than decode with the old one forever + raise ValueError( + f"channel {channel!r} is already bound to {dtype} with" + f" codec {existing.codec!r}; a binding cannot change" + " its codec" + ) + self.inputs.append(_InputBinding(dtype, channel, codec)) # subscribe now, so nothing published before start() is lost: the # queue buffers it and the consumers wait for start anyway logger.info(f"Bind channel {channel!r} ({codec}) to dtype: {dtype}") - self._to_asyncio_task( + return self._to_asyncio_task( self.streamer.subscribe_to_channel, channel, dtype, diff --git a/src/digitaltwin/service/client.py b/src/digitaltwin/service/client.py index 42af91e..52b1281 100644 --- a/src/digitaltwin/service/client.py +++ b/src/digitaltwin/service/client.py @@ -27,7 +27,8 @@ from radical.orbit.client import PluginClient -from ..components import DataType, TypedData +from ..components import DataType, JoinDataType, TypedData +from ..streaming import CODEC_JSON from .wire import ( Package, decode, @@ -239,6 +240,34 @@ def add_agent( twin_id, "add_agent", package, input_dtype, output_dtype, *args, **kwargs )["state"] + def add_input( + self, + twin_id: str, + dtype: DataType, + channel: str, + codec: str = CODEC_JSON, + ) -> str: + """Bind an external channel to one of the twin's input dtypes. + + The producer lives outside the framework and publishes to + `channel`; the twin receives the channel's traffic, decoded by + `codec` (`json` for plain scripts and instruments, `raw` for + bytes, `cloudpickle` only inside one trust domain). Delivery is + the stream's contract: at most once, newest kept under pressure. + The call returns once the binding is live. + """ + + return self._verb(twin_id, "add_input", dtype, channel, codec)["state"] + + def add_data_join(self, twin_id: str, join_dtype: JoinDataType) -> str: + """Register a join: one output event per complete set of inputs. + + `join_dtype` names the member dtypes; components downstream + consume the joined dtype like any other. + """ + + return self._verb(twin_id, "add_data_join", join_dtype)["state"] + def start(self, twin_id: str) -> str: """Start a twin. Starting a running twin is a no-op.""" diff --git a/src/digitaltwin/service/session.py b/src/digitaltwin/service/session.py index 2e14f05..398232d 100644 --- a/src/digitaltwin/service/session.py +++ b/src/digitaltwin/service/session.py @@ -23,9 +23,9 @@ from radical.orbit.plugin_session_base import PluginSession from rhapsody.backends.execution.orbit import OrbitExecutionBackend # type: ignore -from ..components import DataType, TypedData +from ..components import DataType, JoinDataType, TypedData from ..runtime import DTRuntime, RuntimeState -from ..streaming import PubSubClient +from ..streaming import CODEC_JSON, PubSubClient from .wire import Package, check_versions, decode, encode log = logging.getLogger("radical.orbit") @@ -77,6 +77,8 @@ "add_task", "add_investigator", "add_agent", + "add_input", + "add_data_join", "start", "stop", "describe", @@ -443,6 +445,30 @@ async def _verb_add_agent( component = self._instantiate(package, twin) twin.runtime.add_agent(component, input_dtype, output_dtype, *args, **kwargs) + async def _verb_add_input( + self, + twin: TwinInstance, + dtype: DataType, + channel: str, + codec: str = CODEC_JSON, + ) -> None: + # data-only arguments, no Package: the producer lives outside the + # framework, only the binding crosses the wire. The runtime + # validates channel and codec and subscribes at bind time. + subscribed = twin.runtime.add_input(dtype, channel, codec) + + # the verb answers only once the binding is live: a producer + # publishing right after the call must not race the broker + # registration, and a subscribe failure belongs to the caller, + # not to a server-side log alone + if subscribed is not None: + await subscribed + + async def _verb_add_data_join( + self, twin: TwinInstance, join_dtype: JoinDataType + ) -> None: + twin.runtime.add_data_join(join_dtype) + async def _verb_start(self, twin: TwinInstance) -> None: # a running twin is left running: an idempotent retry, not an error twin.runtime.start() diff --git a/src/digitaltwin/streaming.py b/src/digitaltwin/streaming.py index 48f98f8..ae03494 100644 --- a/src/digitaltwin/streaming.py +++ b/src/digitaltwin/streaming.py @@ -926,12 +926,15 @@ async def open( channel: str, codec: str = CODEC_JSON, config: Optional[PubSubConfig] = None, - timeout: Optional[float] = None, + timeout: Optional[float] = CLIENT_CONNECT_TIMEOUT, ) -> "ChannelPublisher": """Connect to a broker and publish to `channel` on it. `config` defaults to the configured broker (environment, else the loopback defaults); its namespace, if it has one, is ignored. + + The connect is bounded by `timeout` -- an external producer + pointed at an unreachable broker should fail fast, not hang. """ config = config or PubSubConfig.resolve() diff --git a/src/digitaltwin/streaming_orbit.py b/src/digitaltwin/streaming_orbit.py index 64a47aa..51fc098 100644 --- a/src/digitaltwin/streaming_orbit.py +++ b/src/digitaltwin/streaming_orbit.py @@ -164,8 +164,14 @@ def _start_runtime(self, timeout: Optional[float]) -> EndpointRuntime: runtime = EndpointRuntime(broker_url=self.broker_url, name=self.name, role="stream") + # ORBIT's start() takes a finite deadline only; the streaming + # contract's "None waits forever" is approximated by a day -- a + # participant that cannot register within that is stuck, not + # waiting. + effective = 86400.0 if timeout is None else timeout + try: - runtime.start(wait=True, timeout=timeout) + runtime.start(wait=True, timeout=effective) # start() returns *silently* when it merely timed out, so the # registration has to be checked explicitly -- otherwise an @@ -174,7 +180,7 @@ def _start_runtime(self, timeout: Optional[float]) -> EndpointRuntime: if not runtime.wait_registered(timeout=0): raise TimeoutError( f"ORBIT broker at {runtime.broker_url} did not register" - f" {self.name} within {timeout}s" + f" {self.name} within {effective}s" ) except BaseException: diff --git a/test/integration/test_dt_orbit_stream.py b/test/integration/test_dt_orbit_stream.py index e97cb2d..a3a4fb3 100644 --- a/test/integration/test_dt_orbit_stream.py +++ b/test/integration/test_dt_orbit_stream.py @@ -15,10 +15,14 @@ import pytest -from digitaltwin.components import TRUTHY, TypedData +from digitaltwin.components import NULL_DTYPE, TRUTHY, DataType, JoinDataType, TypedData from digitaltwin.config import BACKEND_ORBIT from digitaltwin.service import register_user_modules -from digitaltwin.streaming import connect_stream_client +from digitaltwin.streaming import ( + ChannelPublisher, + PubSubConfig, + connect_stream_client, +) import learner_components import twin_components @@ -30,6 +34,7 @@ ECHO_DTYPE, INFERENCE_DTYPE, SENSOR_DTYPE, + JoinSink, OffsetModel, ) @@ -286,3 +291,57 @@ def test_a_persistent_component_publishes_through_the_injected_client( assert seen == sorted(seen), seen orbit_dt.twin_close(twin_id) + + +# --------------------------------------------------------------------------- +# external channels and joins through the client verbs (#30) +# --------------------------------------------------------------------------- + +def test_external_channels_join_through_the_client_verbs(orbit_dt, twin_id): + """The servicified demo shape: two external channels bound with + `add_input`, joined with `add_data_join`, consumed downstream. The + producers are plain `ChannelPublisher`s -- outside the framework, + knowing nothing about twins.""" + + a = DataType("chan-a") + b = DataType("chan-b") + joined = JoinDataType([a, b]) + + chan_a = f"itest/{twin_id[:8]}/a" + chan_b = f"itest/{twin_id[:8]}/b" + + orbit_dt.create_twin(twin_id) + orbit_dt.add_input(twin_id, a, chan_a) + orbit_dt.add_input(twin_id, b, chan_b) + orbit_dt.add_data_join(twin_id, joined) + orbit_dt.add_task(twin_id, orbit_dt.package(JoinSink), joined, NULL_DTYPE) + assert orbit_dt.start(twin_id) == "running" + + async def feed_and_collect(): + config = PubSubConfig(kind=BACKEND_ORBIT, broker_url=ORBIT_BROKER_URL) + collector = await connect_stream_client( + twin_id, backend=BACKEND_ORBIT, broker_url=ORBIT_BROKER_URL) + queue: asyncio.Queue = asyncio.Queue() + pub_a = await ChannelPublisher.open(chan_a, config=config) + pub_b = await ChannelPublisher.open(chan_b, config=config) + + try: + await collector.subscribe_to_dtype(ECHO_DTYPE, queue) + for value in range(3): + await pub_a.publish(value) + await pub_b.publish(value * 10) + return [ + (await asyncio.wait_for(queue.get(), COLLECT_TIMEOUT)).data + for _ in range(3) + ] + finally: + await pub_a.close() + await pub_b.close() + await collector.close() + + seen = asyncio.run(feed_and_collect()) + + # every output is one complete (a, b) pair; joins arrive in order + assert seen == [0, 11, 22], seen + + assert orbit_dt.twin_close(twin_id) == "closed" diff --git a/test/integration/twin_components.py b/test/integration/twin_components.py index 2276aa4..82b3eff 100644 --- a/test/integration/twin_components.py +++ b/test/integration/twin_components.py @@ -121,6 +121,15 @@ async def main_loop(self, runtime, in_data): await runtime.stream.publish(ECHO_DTYPE, in_data.data) +class JoinSink(UtilityTask): + """Terminal for a joined stream: republishes the member sum, so a + client can watch complete join sets arrive over the twin's stream.""" + + async def main_loop(self, runtime, in_data): + await runtime.stream.publish( + ECHO_DTYPE, sum(item.data for item in in_data.data)) + + class CrashingTask(UtilityTask): """Persistent component that dies -- the twin must land in `failed` with the reason visible in `twin_list`.""" diff --git a/test/unit/test_service.py b/test/unit/test_service.py index 42e735b..4d3a814 100644 --- a/test/unit/test_service.py +++ b/test/unit/test_service.py @@ -16,7 +16,12 @@ from fastapi import FastAPI, HTTPException # noqa: E402 from starlette.testclient import TestClient # noqa: E402 -from digitaltwin.components import TRUTHY, DataType, UtilityTask # noqa: E402 +from digitaltwin.components import ( # noqa: E402 + TRUTHY, + DataType, + JoinDataType, + UtilityTask, +) from digitaltwin.runtime import DTRuntime # noqa: E402 from digitaltwin.service.plugin import UI_ASSETS, PluginDT # noqa: E402 from digitaltwin.service.session import DTSession, TwinInstance # noqa: E402 @@ -902,3 +907,106 @@ async def test_endpoint_loss_spares_pool_backed_roles(): assert session.endpoints_lost({"hpc1", "exsitu", "pool"}) == () assert not session._lost + + +# --------------------------------------------------------------------------- +# graph verbs: external inputs and joins (#30) +# --------------------------------------------------------------------------- + +class _BindingStream(_FakeStream): + """A `_FakeStream` that also accepts channel subscriptions.""" + + def __init__(self): + self.subscribed = [] + + async def subscribe_to_channel(self, channel, dtype, queue, codec): + self.subscribed.append((channel, dtype, codec)) + + +async def test_add_input_binds_a_channel_through_the_verb(): + """`add_input` carries data only -- dtype, channel, codec -- and the + binding lands in the runtime exactly as a direct call would leave it.""" + + session = DTSession("s1") + twin = _running_twin(session, "t1") + twin.runtime.streamer = _BindingStream() + + x = DataType("x") + await session.twin_call( + "t1", "add_input", + encode({"args": (x, "lab/raw")}), version_stamp()) + + binding = twin.runtime.inputs[0] + assert (binding.dtype, binding.channel, binding.codec) == (x, "lab/raw", + "json") + # the verb answered, so the subscription is already live -- no settling + assert twin.runtime.streamer.subscribed == [("lab/raw", x, "json")] + assert twin.summary()["calls"] == {"add_input": 1} + + await twin.close() + + +async def test_add_input_rejects_a_bad_channel_as_client_error(): + """The runtime's own refusal (reserved prefix) surfaces as a 409, + the same contract every other graph verb has.""" + + session = DTSession("s1") + twin = _running_twin(session, "t1") + twin.runtime.streamer = _BindingStream() + + with pytest.raises(HTTPException) as raised: + await session.twin_call( + "t1", "add_input", + encode({"args": (DataType("x"), "dt/oops")}), version_stamp()) + + assert raised.value.status_code == 409 + assert twin.runtime.inputs == [] + + await twin.close() + + +async def test_add_data_join_registers_through_the_verb(): + """One joined dtype, consumable downstream like any other.""" + + session = DTSession("s1") + twin = _running_twin(session, "t1") + + a, b = DataType("a"), DataType("b") + joined = JoinDataType([a, b]) + await session.twin_call( + "t1", "add_data_join", encode({"args": (joined,)}), version_stamp()) + + assert joined in twin.runtime.join_components + assert twin.summary()["calls"] == {"add_data_join": 1} + + await twin.close() + + +async def test_add_input_refuses_a_codec_change_on_a_bound_channel(): + """The stream client dedupes on (channel, dtype): a re-bind with a + different codec would be recorded yet never applied, so it is + refused instead of silently decoding with the old codec forever.""" + + session = DTSession("s1") + twin = _running_twin(session, "t1") + twin.runtime.streamer = _BindingStream() + + x = DataType("x") + await session.twin_call( + "t1", "add_input", encode({"args": (x, "lab/raw")}), version_stamp()) + + # the identical re-bind is an idempotent no-op + await session.twin_call( + "t1", "add_input", encode({"args": (x, "lab/raw")}), version_stamp()) + assert len(twin.runtime.inputs) == 1 + + with pytest.raises(HTTPException) as raised: + await session.twin_call( + "t1", "add_input", + encode({"args": (x, "lab/raw", "raw")}), version_stamp()) + + assert raised.value.status_code == 409 + assert "codec" in raised.value.detail + assert len(twin.runtime.inputs) == 1 + + await twin.close() diff --git a/test/unit/test_streaming_orbit.py b/test/unit/test_streaming_orbit.py index f385a74..631c28b 100644 --- a/test/unit/test_streaming_orbit.py +++ b/test/unit/test_streaming_orbit.py @@ -518,3 +518,33 @@ async def test_participant_names_are_unique(): # ... unless the caller insists assert OrbitPubSubBackend(name="fixed").name == "fixed" + + +def test_a_none_connect_timeout_reaches_orbit_as_a_finite_deadline( + monkeypatch): + """ORBIT's `start()` takes a finite float only; the streaming + contract's `None` ("wait forever") must arrive as the documented + day-long approximation, not as `None` (`monotonic() + None` dies).""" + + recorded = {} + + class _Recording: + def __init__(self, broker_url=None, name=None, role=None): + pass + + def start(self, wait=True, timeout=None): + recorded["timeout"] = timeout + + def wait_registered(self, timeout=0): + return True + + monkeypatch.setattr( + "digitaltwin.streaming_orbit.EndpointRuntime", _Recording) + + backend = OrbitPubSubBackend() + backend._start_runtime(None) + + assert recorded["timeout"] == 86400.0 + + backend._start_runtime(5.0) + assert recorded["timeout"] == 5.0