DTaaS M3: ORBIT-pubsub stream backend — data plane inside the token domain (closes R7)

.github/workflows/ci.yml

@@ -32,12 +32,13 @@ jobs:
         run: >-
           pip install "radical.asyncflow==0.5.1" "radical.orbit==0.5.0"
           "rhapsody-py[telemetry] @ git+https://github.com/radical-cybertools/rhapsody@e491cd2"
+          "rose @ git+https://github.com/radical-cybertools/ROSE@64330d9cb43c3e13ca67daf0d8ae84a2ae6c3f17"
 
       # src/ layout: pytest tests the *installed* package, never the
       # working tree.  Always install before testing -- a stale install
       # will silently pass against old code.
       - name: Install digitaltwin (test + service extras)
-        run: pip install ".[test,service]" pytest-timeout
+        run: pip install ".[test,service,learn]" pytest-timeout
 
       # --continue-on-collection-errors: one broken test module (see the
       # PR description) must not hide the rest of the suite behind an
@@ -71,12 +72,13 @@ jobs:
         run: >-
           pip install "radical.asyncflow==0.5.1" "radical.orbit==0.5.0"
           "rhapsody-py[telemetry] @ git+https://github.com/radical-cybertools/rhapsody@e491cd2"
+          "rose @ git+https://github.com/radical-cybertools/ROSE@64330d9cb43c3e13ca67daf0d8ae84a2ae6c3f17"
 
       # src/ layout: pytest tests the *installed* package, never the
       # working tree.  Always install before testing -- a stale install
       # will silently pass against old code.
       - name: Install digitaltwin (test + service extras)
-        run: pip install ".[test,service]" pytest-timeout
+        run: pip install ".[test,service,learn]" pytest-timeout
 
       # The ORBIT broker needs a self-signed TLS cert/key and a shared
       # ingress token at the default ~/.radical/orbit location -- see

README.md

@@ -5,7 +5,7 @@ Main set of features implemented:
 - Utility Tasks
 - Persistent Tasks
 - Callbacks
-- Simple ZMQ pubsub backend
+- Two pubsub backends: ZMQ, and ORBIT eventing
 - Graph builder
 - Convert to a Python Package
 - Several Tests / Examples
@@ -69,7 +69,57 @@ are cloudpickled, so anyone who can reach the broker ports can execute code
 in every subscriber.  A non-loopback bind needs an explicit configuration
 and a private/firewalled network.  External channels are decoded with the
 codec their binding names: `json` (the default) and `raw` are safe to
-accept from a producer you do not control, `cloudpickle` is not.
+accept from a producer you do not control, `cloudpickle` is not.  The
+demos are the reason the ZMQ backend exists; anything beyond a laptop
+should be on the ORBIT one below.
+
+## Choosing a data plane
+
+`DT_STREAM_BACKEND` picks which transport carries the twins' streams.  It
+is a **deployment-time** choice, resolved once where the framework runs;
+no client and no session can ask for a different one.  Nothing above
+`PubSubBackend` -- not `DTRuntime`, not a component, not the injected
+`RuntimeAPI.stream` client -- knows which is in use.
+
+| `DT_STREAM_BACKEND` | transport | ports it opens | use |
+|---------------------|-----------|----------------|-----|
+| `zmq` (default)     | the framework's own XSUB/XPUB broker | two, unauthenticated, loopback by default | local, demos, the two-terminal loop |
+| `orbit`             | ORBIT eventing (`radical.orbit`)     | **none** | anything shared, and everything in production |
+
+```sh
+# a service deployment with the data plane inside the token domain
+DT_STREAM_BACKEND=orbit radical-orbit-broker.py --plugins default,dt
+```
+
+An external subscriber joins the same way -- as an ORBIT participant, so
+it needs the broker URL and the token, and no addresses at all:
+
+```python
+from digitaltwin.streaming import connect_stream_client
+
+stream = await connect_stream_client(twin_id, backend='orbit')
+await stream.subscribe_to_dtype(ECHO, queue)
+```
+
+**Payload ceiling**: an ORBIT frame is capped at 4 MiB, so a single
+stream message must cloudpickle to less than that (64 KiB of the budget
+is reserved for the envelope).  Oversized payloads raise a clear
+`ValueError` at `publish` -- ORBIT itself would drop the frame with
+nothing but a log line, which on a days-long twin is indistinguishable
+from a stalled stream.  The ZMQ backend has no such ceiling; a twin meant
+to run on either should stay well under it.  Chunk large artifacts, or
+stream a reference and move the bytes with the staging plugin.
+
+**Semantics** are the same on both: at-most-once, with bounded
+drop-oldest queues (broker-side, and again on the hop into the host
+loop).  That *is* the DT conflation contract, so nothing above the
+backend adds a second one -- a slow consumer loses samples rather than
+memory, and never backpressures a producer.  Loss is visible as a gap in
+the broker-assigned sequence numbers.  ORBIT's `replay` plugin would give
+late joiners history; it is deliberately not integrated in v1.
+
+`perf/bench_streams.py` measures what the choice costs: about a
+millisecond per stream hop on loopback.
 
 ## Running it as a service (the `dt` ORBIT plugin)
 
@@ -195,17 +245,43 @@ again, so a `twin_create` after it fails immediately with `engine
 `ready` and stalling.  `unregister_session`, then build the session and
 its twins again.
 
-### Binding policy for the service (R7)
+### The data plane and its trust boundary (R7)
 
-The plugin runs its own DT stream broker, embedded, one per plugin and
-shared by every twin.  **It binds to loopback on a random port by
-default, and that default is the safe one**: its payloads are
-cloudpickled, so anyone who can reach the XSUB/XPUB ports gets code
-execution in every subscriber -- weaker than the token-authenticated
-ORBIT channel around it.
+The DT streams carry cloudpickled payloads.  That is accepted -- the
+service already executes client-shipped component classes, and both sit
+inside ORBIT's single-token trust domain (risk R4).  What was *not*
+acceptable is where those payloads used to travel: a pair of ZMQ ports
+that authenticate nobody, so anyone who could reach them got code
+execution in every subscriber, no token required.  The data plane was
+weaker than the control plane wrapped around it.
+
+**`DT_STREAM_BACKEND=orbit` closes that gap**, and a production
+deployment must use it:
+
+```sh
+DT_STREAM_BACKEND=orbit radical-orbit-broker.py --plugins default,dt
+```
 
-A non-loopback bind is possible (`DT_STREAM_PUB_ADDR` /
-`DT_STREAM_SUB_ADDR` on the service host) but requires a deliberate
-decision *and* a firewalled or private network.  Until the data plane
-moves inside ORBIT's authenticated channel, do not expose those ports --
-including in demos.
+The twins' streams become ORBIT events on the same token-authenticated
+WebSocket star as every other call, under one `dt_stream` plugin
+namespace.  The embedded ZMQ broker is then **never started** -- the
+service opens no data-plane port at all, and there is nothing left to
+firewall.  The payloads are still cloudpickle; what changed is that
+reaching them now requires the same token as calling `twin_create`.
+Reviewers can check the guarantee directly: the plugin host has no child
+processes, and `admin/sessions` reports `{"stream_broker": {"backend":
+"orbit"}}`.
+
+Two things this does *not* do.  It does not make the payloads safe to
+receive from an untrusted party -- per-tenant auth is post-v1, so
+everything inside the token domain is still mutually trusting.  And it
+does not remove the 4 MiB frame cap, which the ZMQ backend did not have
+(see "Choosing a data plane").
+
+**With the `zmq` backend the old mitigations still apply, in full.** The
+plugin runs its own DT stream broker, embedded, one per plugin and shared
+by every twin.  It binds to loopback on a random port by default, and
+that default is the safe one.  A non-loopback bind is possible
+(`DT_STREAM_PUB_ADDR` / `DT_STREAM_SUB_ADDR` on the service host) but
+requires a deliberate decision *and* a firewalled or private network.  Do
+not expose those ports -- including in demos.

perf/README.md

@@ -29,6 +29,40 @@ Routing all user compute through the Rhapsody abstraction therefore
 costs single-digit milliseconds per sequential prediction and wins by an
 order of magnitude under concurrency.
 
+## `bench_streams.py` — stream latency, ZMQ vs ORBIT data plane
+
+One publish awaited until the subscriber's queue hands it back: the shape
+of every hop in a twin's graph.  Both rows go through the same
+`PubSubClient`, so the only difference is the backend (M3).
+
+```sh
+# the zmq row starts its own embedded broker; the orbit row needs a live one
+python perf/bench_streams.py both --broker https://127.0.0.1:8031
+```
+
+Loopback, one host, bare-int payloads (2026-08-15):
+
+| data plane                  | p50     | p99     | burst          |
+|-----------------------------|---------|---------|----------------|
+| zmq, embedded broker        | 0.98 ms | 1.32 ms | 20 200 msg/s   |
+| orbit eventing              | 2.09 ms | 2.60 ms |  4 300 msg/s   |
+
+With 64 KiB payloads (`--payload 65536`): 1.18 ms / 3.56 ms p50, and
+6 500 vs 1 150 msg/s in burst.
+
+So the ORBIT data plane costs roughly **1 ms per stream hop** and about a
+quarter of the burst throughput, in exchange for the security property of
+M3: the payloads ride the token-authenticated WebSocket star and the
+deployment opens no unauthenticated ports (risk R7).  Against the ~20 ms
+of a single in-situ prediction (the row above), that is noise.
+Informational, not a gate.
+
+The extra hop is structural: ZMQ's XSUB/XPUB proxy forwards a frame
+between two sockets, while an ORBIT event is packed, sent to the broker,
+stamped with a `seq`, fanned out, and handed across a thread boundary
+into the host loop.  Payload size hurts the ORBIT row more because the
+frame is msgpacked around the pickle.
+
 ## `streaming_learner_perf.py`, `plot_streaming_perf.py`
 
 Throughput of the streaming active learner (ROSE); unrelated to the

perf/bench_streams.py

@@ -0,0 +1,177 @@
+"""Stream latency: the ZMQ data plane vs the ORBIT one.
+
+One publish, awaited until the subscriber's queue hands it back -- the
+shape of every hop in a twin's graph (a persistent component publishes a
+dtype, the runtime consumes it).  Both rows go through the *same*
+`PubSubClient`, so the difference is the backend and nothing else.
+
+The ZMQ row starts its own embedded broker on a random loopback port,
+exactly as the service does.  The ORBIT row needs a live broker::
+
+    # against the integration stack (test/integration/conftest.py has one)
+    python perf/bench_streams.py both --broker https://127.0.0.1:8031
+
+Informational, not a gate: the ORBIT row buys the security property of
+milestone M3 (no unauthenticated ports) and pays a WebSocket round trip
+through the broker for it.
+"""
+
+import argparse
+import asyncio
+import os
+import sys
+import time
+
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).resolve().parent))
+
+from bench_insitu import report  # noqa: E402
+
+from digitaltwin import DataType, ZMQ_BrokerProcess  # noqa: E402
+from digitaltwin.config import BACKEND_ORBIT, BACKEND_ZMQ  # noqa: E402
+from digitaltwin.streaming import connect_stream_client  # noqa: E402
+
+SAMPLE = DataType("sample")
+
+N_WARM = 20
+N_MEAS = 200
+N_BURST = 200
+
+
+SETTLE_TIMEOUT = 30.0
+DELIVER_TIMEOUT = 10.0
+
+
+QUIET_WAIT = 0.25
+
+
+async def settle(client, queue, message) -> None:
+    """Publish until something comes back, then wait for quiet.
+
+    Neither backend acknowledges a subscription -- ZMQ's SUBSCRIBE and
+    ORBIT's `subscribe` frame are both fire-and-forget -- so the only
+    honest barrier is a message that made the round trip.
+
+    Draining until the queue *stays* empty matters as much as the barrier
+    itself: a `queue.empty()` check would leave any barrier message still
+    in flight to arrive during the measurement, where it would pair with
+    the wrong publish and shift every latency after it.
+    """
+
+    deadline = time.perf_counter() + SETTLE_TIMEOUT
+
+    while time.perf_counter() < deadline:
+        await client.publish(SAMPLE, message)
+        try:
+            await asyncio.wait_for(queue.get(), QUIET_WAIT)
+            break
+        except TimeoutError:
+            continue
+    else:
+        raise TimeoutError(f"no message came back within {SETTLE_TIMEOUT}s")
+
+    while time.perf_counter() < deadline:
+        try:
+            await asyncio.wait_for(queue.get(), QUIET_WAIT)
+        except TimeoutError:  # nothing left in flight
+            return
+
+    raise TimeoutError(f"the stream never went quiet within {SETTLE_TIMEOUT}s")
+
+
+async def measure(client, label: str, n_meas: int, n_burst: int,
+                  payload: int) -> None:
+    """Publish -> deliver round trips through one stream client."""
+
+    queue: asyncio.Queue = asyncio.Queue()
+    await client.subscribe_to_dtype(SAMPLE, queue)
+
+    message = b"x" * payload if payload else 0
+
+    await settle(client, queue, message)
+
+    for _ in range(N_WARM):
+        await client.publish(SAMPLE, message)
+        await asyncio.wait_for(queue.get(), DELIVER_TIMEOUT)
+
+    latencies = []
+    for _ in range(n_meas):
+        start = time.perf_counter()
+        await client.publish(SAMPLE, message)
+        await asyncio.wait_for(queue.get(), DELIVER_TIMEOUT)
+        latencies.append(time.perf_counter() - start)
+
+    report(label, latencies)
+
+    # burst: how fast the data plane drains a producer that does not wait.
+    # Both backends drop the oldest when a queue overruns, so the count
+    # that arrives is part of the measurement.
+    start = time.perf_counter()
+    for _ in range(n_burst):
+        await client.publish(SAMPLE, message)
+
+    received = 0
+    while received < n_burst:
+        try:
+            await asyncio.wait_for(queue.get(), 5.0)
+        except TimeoutError:
+            break
+        received += 1
+
+    elapsed = time.perf_counter() - start
+    print(f"{label:28s} {received}/{n_burst} burst messages in "
+          f"{elapsed * 1000:.1f}ms ({received / elapsed:.0f} msg/s)")
+
+
+async def bench_zmq(args) -> None:
+    broker = ZMQ_BrokerProcess()
+    await broker.start()
+
+    try:
+        client = await connect_stream_client(
+            "bench-zmq", *broker.get_connection_str(), backend=BACKEND_ZMQ
+        )
+        try:
+            await measure(client, "zmq (embedded broker)", args.messages,
+                          args.burst, args.payload)
+        finally:
+            await client.close()
+    finally:
+        await broker.stop()
+
+
+async def bench_orbit(args) -> None:
+    client = await connect_stream_client(
+        "bench-orbit", backend=BACKEND_ORBIT, broker_url=args.broker
+    )
+    try:
+        await measure(client, "orbit eventing", args.messages, args.burst,
+                      args.payload)
+    finally:
+        await client.close()
+
+
+async def main() -> None:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("which", nargs="?", default="both",
+                        choices=["zmq", "orbit", "both"])
+    parser.add_argument("--broker", default=os.environ.get(
+        "RADICAL_ORBIT_BROKER_URL"),
+        help="ORBIT broker URL (default: ORBIT's own resolution)")
+    parser.add_argument("--messages", type=int, default=N_MEAS)
+    parser.add_argument("--burst", type=int, default=N_BURST)
+    parser.add_argument("--payload", type=int, default=0,
+                        help="payload size in bytes (0: a bare int)")
+
+    args = parser.parse_args()
+
+    if args.which in ("zmq", "both"):
+        await bench_zmq(args)
+
+    if args.which in ("orbit", "both"):
+        await bench_orbit(args)
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

src/digitaltwin/__init__.py

@@ -13,7 +13,7 @@
     WindowDataType,
     WindowedTypeData,
 )
-from .config import stream_addresses
+from .config import stream_addresses, stream_backend
 from .runtime import DTRuntime, RuntimeAPI, RuntimeState
 from .streaming import (
     CODEC_CLOUDPICKLE,
@@ -56,4 +56,5 @@
     "ZMQ_PS_Client",
     "connect_stream_client",
     "stream_addresses",
+    "stream_backend",
 ]