DTaaS M3: ORBIT-pubsub stream backend — data plane inside the token domain (closes R7) - #2
Merged
Conversation
The DT streams carry cloudpickled payloads over ZMQ ports that nobody authenticates: reaching them is code execution in every subscriber (plan risk R7). This puts the same payloads inside ORBIT's token-authenticated star instead, behind the backend seam M0 built, so nothing above it notices. `OrbitPubSubBackend` publishes with `EndpointRuntime.send_notification` and subscribes with `register_callback`, one registration per DT topic under a single `dt_stream` plugin namespace -- ORBIT matches topics by exact equality, which is what DT topics already are. It owns one participant connection per twin (rhapsody's `OrbitExecutionBackend` pattern, loopback wrinkle and all), or rides an injected one. Frames cross from the runtime's callback thread to the host loop through a bounded drop-oldest inbox -- the broker's own discipline, and the conflation semantics the data plane is specified with. Oversized payloads raise at publish, because ORBIT would drop them with nothing but a log line. The two backends had grown the same receive loop, subscriber registry and closed/running state, so those move up into `PubSubBackend`. `DT_STREAM_BACKEND` selects the transport at deployment time. Under 'orbit' the plugin's embedded ZMQ broker is never started at all -- the `connect_stream` seam is the only caller of `stream_addresses`, so that is structural rather than remembered. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
The stand-in reproduces the three ORBIT behaviours the backend is written against: exact-match subscriptions, fan-out that does *not* exclude the sender, and callbacks arriving on a foreign thread. The first of those is load-bearing for DT -- a twin's runtime consumes the dtypes its own persistent components publish, so a backend that never got its own events back would deliver nothing at all. Covers namespacing isolation on the second backend (same dtype label, two twins, one broker), unsubscribe and re-subscribe, one wire subscription per topic however many local subscribers, the payload ceiling, drop- oldest under a stalled loop, and teardown: the participant is stopped, an injected one is not, and a runtime that cannot register is not left behind. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
A second deployment on the next port, started with DT_STREAM_BACKEND=orbit -- separate because the transport is a deployment-time choice and no client can ask for it. The tests run whole twins on it: sensor to investigator to sink, the M2 learner retraining off the twin's input stream, two twins with the same dtype label staying apart, inference unaffected, churn leaking no participants. The subscriber they attach is itself an ORBIT participant, so even the observation path opens no port -- where the ZMQ suite has to go and ask the service for addresses. The R7 assertion is the direct one: the embedded stream broker is a spawned subprocess, so "none was started" is "the plugin host has no children", checked after a twin has actually streamed. Plus the oversized-payload refusal, which is the one place the 4 MiB frame cap is visible to a user. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
`bench_streams.py` times publish -> deliver through the same `PubSubClient` on both backends, so the difference is the transport and nothing else. Loopback: 0.84 ms p50 on ZMQ, 1.94 ms on ORBIT, and about a quarter of the burst throughput. Roughly a millisecond per hop for the security property -- noise against the ~20 ms of a single in-situ prediction. Numbers in the perf README, informational, not a gate. Neither backend acknowledges a subscription, so the harness opens with a round trip it retries until one comes back; without that barrier the first publishes are simply lost and the measurement never starts. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
The framework's own pubsub ports authenticate nobody, so the data plane was weaker than the control plane wrapped around it: reaching those ports was code execution in every subscriber, token-free (risk R7). `DT_STREAM_BACKEND=orbit` puts the same payloads on the token- authenticated star instead, and starts no ZMQ broker at all -- there is then nothing left to firewall. Says so plainly, including the two things it does not do (per-tenant auth is still post-v1; the 4 MiB frame cap is new and the ZMQ backend has none), how a reviewer checks the guarantee, and that the ZMQ mitigations remain in force for the deployments that keep it. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
`_await_running` parks a publish or subscribe that arrives before connect() finished. If the client is closed while somebody waits there, the wait ends on a backend that has already let its transport go -- so re-check, and raise the ordinary closed-client error instead of failing somewhere further in. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
- The docstring claimed two bounded queues on the receive path; there are three. The middle one is ORBIT's own `CallbackDispatcher`, and it is the odd one out -- it drops the *newest* frame, not the oldest. Name it, and log its counter alongside ours at close, so a twin that lost samples does not have to be inferred from `seq` gaps in somebody else's log. - `_await_running` parked a caller that arrived before connect. If the connect was then abandoned, `close()` merely cleared the flag and the waiter kept waiting for a connection that would never come. Closing now sets the event before clearing it: the waiters wake, hit `_check_open`, and get the ordinary closed-client error. Tested for both verbs that can park. - The zmq branch of `connect_stream` re-read the environment per twin, which could contradict the choice the plugin resolved once at construction. Name the backend. - `frame_cap()` reaches into a private `EndpointRuntime` attribute; say so, and that the fix belongs upstream. - The benchmark's barrier drained with an `empty()` check, which leaves a barrier message still in flight to arrive mid-measurement and pair with the wrong publish. Wait for quiet instead. Numbers re-measured. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
The M3 data plane added broker_url to PubSubConfig; the M0 round-trip test compares the dict shape exactly and follows. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
A channel's payload is encoded by the codec its binding names, so the transport has to hand those bytes over untouched -- the `raw` flag the seam already carries. The ZMQ backend honoured it; the ORBIT one did not, so a twin binding a channel on that data plane failed at the first subscribe. The set of raw topics is the same notion in both backends, so it moves up to PubSubBackend where the rest of the shared receive machinery already lives. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
The devel merge changed `PubSubBackend.unsubscribe` from `async def` to
`def`, and `PubSubClient` now calls it without `await`. Two spots on
this branch were still on the old contract:
* `OrbitPubSubBackend.unsubscribe` still declared `async def` -- its
body is entirely synchronous -- so the client's plain call created a
coroutine and dropped it, and the unsubscribe silently never
happened ("coroutine ... was never awaited" in the test run).
* the test awaited `unsubscribe_dtype`, which now returns None, so it
died on a TypeError before it could even observe the first problem.
Both are the same mismatch, both fixed by dropping the keyword.
test_streaming_orbit: 17 passed (was 1 failed, 16 passed). The two
remaining unit failures on this branch (`kind == "generic"` and the
describe() shape) are devel regressions, fixed on fix/devel-regressions;
they converge when that merges and devel flows down.
Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
BenCarter44
approved these changes
Aug 19, 2026
BenCarter44
left a comment
BenCarter44
left a comment
Collaborator
There was a problem hiding this comment.
Looks great! Looked through it all and makes sense to me.
Ready to merge once #1 is merged.
This was referenced Aug 23, 2026
M2/M3 bring integration tests importing `rose`. PyPI's `rose` is an unrelated version-string helper, so the real one is pinned like the other radical deps, and the learn extra installs. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
2 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Milestone M3 (final) of the DTaaS v1 plan (
docs/dtaas-v1-plan.md): the DT data plane moves onto ORBIT's token-authenticated eventing — closing the plan's R7 security risk for service deployments.Stacked on #5 (M2) → #4 (M1) → #3 (M0) — review the commits from the M2 head onward, or merge in order.
What's in here
OrbitPubSubBackend(digitaltwin/streaming_orbit.py, lazy-imported): a second stream backend behind the abstraction M0 built. Publish = one participanteventframe per message (EndpointRuntime.send_notification, a public bare-participant path); subscribe = exact-topic broker-side filtering; delivery = bounded drop-oldest inbox bridged onto the host loop — matching the DT conflation contract, not adding a second layer (all three bounded hops documented, drops counted and reported at close).DT_STREAM_BACKEND=zmq|orbit, resolved once at plugin construction (invalid value fails the deployment), applied in a single choke point — underorbitthe embedded ZMQ broker subprocess is structurally never spawned.admin/sessionsreports the active backend.perf/bench_streams.py): stream hop p50 0.98 ms (zmq) vs 2.09 ms (orbit) — ~1.1 ms per hop for the token-domain property, noise against a ~20 ms in-situ prediction.Internally reviewed (approve; all findings applied). This completes the v1 milestone series: M0 hardening → M1 service plugin → M2 ROSE ex-situ learning → M3 secured data plane.
🤖 Generated with Claude Code