feat(bridge): add Python bridge runtime (hub, interface, host runner) - #3696
feat(bridge): add Python bridge runtime (hub, interface, host runner)#3696ryanmelt wants to merge 4 commits into
Conversation
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Codecov Report✅ All modified and coverable lines are covered by tests. Additional details and impacted files@@ Coverage Diff @@
## pr2-bridge-models #3696 +/- ##
=====================================================
- Coverage 79.30% 78.66% -0.64%
=====================================================
Files 885 890 +5
Lines 65367 66343 +976
Branches 2537 2591 +54
=====================================================
+ Hits 51838 52192 +354
- Misses 12856 13482 +626
+ Partials 673 669 -4
Flags with carried forward coverage won't be shown. Click here to find out more. ☔ View full report in Codecov by Harness. 🚀 New features to boost your workflow:
|
When a host interface fails to open its device (e.g. a missing USB HID device: "HIDAPI Device ... Not Found"), the host parks without ever reporting CONNECTED, so the CONNECTED-gated detection never fired and the COSMOS bridge_interface stayed connected. The host now reports its `desired` state (pushed immediately on change), and bridge_interface returns nil on a desired True->False transition, so COSMOS disconnects and reconnects. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
iroh's IrohError keeps its detail behind .message() (str()/traceback show only the class name), so bridge crash/error logs were opaque. Add _iroh_error_detail and route all iroh-error logs in the hub, COSMOS interface, and host runner through it. Also downgrade the hub's inbound accept/negotiation failures from warn to debug: non-iroh QUIC clients probing the open UDP port fail ALPN/TLS negotiation (NoApplicationProtocol / WebPKI UnknownIssuer) and spammed the log. Legit peers pin the node key via the ticket and never hit this; unauthorized peers that do negotiate still log at warn. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
![github-code-quality[bot]](https://avatars.githubusercontent.com/u/9919?s=60&v=4){kind=small}
mcosgriff
left a comment
mcosgriff
left a comment
There was a problem hiding this comment.
Probable good to get tests in there
| if channel.startswith(CTRL_CHANNEL_PREFIX): | ||
| ctrl_name = channel[len(CTRL_CHANNEL_PREFIX) :].decode("utf-8", "replace") | ||
|
|
||
| partner = self._waiting.pop(channel, None) |
There was a problem hiding this comment.
rendezvous pairs any two legs, no side check, no liveness check
_waiting keyed only by (both stream/NAME and host/NAME map to b"NAME"). is_host stored at line 576 but never used for validation — only for ctrl tap direction. Three failures fall out:
- Parked entry never removed when its connection dies. COSMOS _establish_data times out at 30s and calls _close_data(), but hub's parked task blocks in wait_for(future, 300) — nothing watches conn.closed(). Entry rots in _waiting up to 300s.
- COSMOS reconnect (~5s InterfaceMicroservice retry) pops that dead COSMOS entry and pairs COSMOS↔COSMOS. Pump instantly EOFs, so the fresh leg never parks to wait for the real host. Steady-state flapping when host is down.
- Line 576 overwrites an existing _waiting[channel] without closing it — orphaned connection + task blocked for PAIR_TIMEOUT.
- except asyncio.TimeoutError (579) doesn't catch CancelledError, so shutdown leaves entries too. Pop in finally.
Fix: key on channel, require partner.is_host != is_host, pop in finally, and race the parked wait_for against conn.closed().
| if not self._started: | ||
| self._loop = asyncio.new_event_loop() | ||
| self._thread = threading.Thread(target=self._run_loop, daemon=True) | ||
| self._thread.start() | ||
| asyncio.run_coroutine_threadsafe(self._startup(), self._loop).result(self.connect_timeout) | ||
| self._ctrl_task = asyncio.run_coroutine_threadsafe(self._start_control(), self._loop).result(5) | ||
| self._started = True |
There was a problem hiding this comment.
thread + event-loop leak per failed connect
if not self._started:
self._loop = asyncio.new_event_loop()
self._thread = threading.Thread(target=self._run_loop, daemon=True); self._thread.start()
asyncio.run_coroutine_threadsafe(self._startup(), self._loop).result(self.connect_timeout)
...
self._started = True
_startup raises (iroh bind failure, Redis error in BridgeInterfaceModel.create) or .result() times out ⇒ _started stays False but thread already spawned running run_forever. Next retry builds a new loop+thread and orphans the old. InterfaceMicroservice retries forever ⇒ unbounded thread growth. Tear down loop/thread in an except before re-raising.
| # reconnect logic retries. The +5 lets the inner handshake timeout fire | ||
| # (and clean up the tunnel) before this outer wait gives up. | ||
| self._want_connected = True | ||
| self._send_command("connect") |
There was a problem hiding this comment.
vs :214 — _host_attempt_seen reset happens after the connect command
connect() sends "connect" (177), then _establish_data clears _host_attempt_seen = False (214). Host's desired=True status can land in between and get erased. Host then parks (device open fails) ⇒ desired=False arrives with _host_attempt_seen False ⇒ detection never fires, and subsequent heartbeats stay False so it never fires later either. That's the exact case commit 9f6b23c added. Only the data-leg EOF saves it — the path that commit says isn't reliable. Move the reset above _send_command("connect").
| self.streams = streams | ||
| with contextlib.suppress(Exception): | ||
| endpoint.set_alpns(self._build_alpns()) |
There was a problem hiding this comment.
failed set_alpns becomes permanent
self.streams = streams
with contextlib.suppress(Exception):
endpoint.set_alpns(self._build_alpns())State committed before the call. If set_alpns throws, next iteration sees streams == self.streams and returns early forever — new bridged interfaces silently unreachable until hub restart, no log. Assign self.streams only after success, and log the exception.
| def _ensure_port(self, model): | ||
| """Return this bridge's fixed UDP port, assigning one from the published | ||
| range if it doesn't have one yet. The port is persisted on the model and | ||
| reused across restarts so the host can always reach the hub at | ||
| 127.0.0.1:<port>. Ports must be unique across every bridge sharing this | ||
| operator container, so the lowest port not claimed by any other bridge | ||
| (in any scope) is chosen and persisted immediately.""" | ||
| if getattr(model, "port", None): | ||
| return int(model.port) | ||
|
|
||
| used = set() | ||
| for scope in ScopeModel.names(): | ||
| for name in BridgeModel.names(scope): | ||
| if scope == self.scope and name == self.bridge_name: | ||
| continue | ||
| other = BridgeModel.get_model(name, scope=scope) | ||
| if other and getattr(other, "port", None): | ||
| used.add(int(other.port)) | ||
|
|
||
| for candidate in range(BRIDGE_PORT_BASE, BRIDGE_PORT_BASE + BRIDGE_PORT_COUNT): | ||
| if candidate not in used: | ||
| model.port = candidate | ||
| model.create(force=True) | ||
| return candidate | ||
|
|
||
| raise RuntimeError( | ||
| f"No free bridge port in {BRIDGE_PORT_BASE}-{BRIDGE_PORT_BASE + BRIDGE_PORT_COUNT - 1}; " | ||
| "increase OPENC3_BRIDGE_PORT_COUNT and the published range in compose.yaml" | ||
| ) |
There was a problem hiding this comment.
_ensure_port TOCTOU
Two hubs starting concurrently both scan used, both pick the lowest free port, both persist it, second Endpoint.bind fails. No lock, no retry-on-bind-failure. Bind first then persist, or retry the next candidate on AddrInUse.
| async def _read_request(self, recv): | ||
| """Read a full request from a bi-stream until the peer finishes writing. | ||
| A single `read()` can return only a partial payload (especially over a | ||
| relay, where data arrives in smaller chunks), so loop to EOF to avoid | ||
| truncated/undecodable JSON.""" | ||
| data = b"" | ||
| with contextlib.suppress(Exception): | ||
| while not self.cancel_thread: | ||
| chunk = await recv.read(PUMP_CHUNK_BYTES) | ||
| if not chunk: | ||
| break | ||
| data += bytes(chunk) | ||
| return data |
There was a problem hiding this comment.
_read_request has no timeout
Loops to EOF with only cancel_thread as an exit. api/enroll is deliberately pre-auth (bootstrap), so any peer that can negotiate it holds a handler task and connection open indefinitely. Wrap in asyncio.wait_for and cap len(data).
|
| return f"{type(error).__name__}: {detail}" | ||
| except Exception: | ||
| # Error-detail extraction is best effort; use str(error) below. | ||
| detail = None |
| return f"{type(error).__name__}: {detail}" | ||
| except Exception: | ||
| # Error-detail extraction is best effort; use str(error) below. | ||
| detail = None |
| return f"{type(error).__name__}: {detail}" | ||
| except Exception: | ||
| # Error-detail extraction is best effort; use str(error) below. | ||
| detail = None |
|
|
||
| # 4. Now connect the device and start pumping raw bytes. | ||
| interface = self.build_interface() | ||
| interface.connect() |
There was a problem hiding this comment.
interface.connect() executes arbitrary device/plugin code synchronously on the event loop responsible for control commands, status updates, and shutdown. A blocking serial/TCP/device open prevents disconnects and shutdown indefinitely; interface.disconnect() at line 434 has the same issue. Run both in an executor/thread with an appropriate timeout.
| if self.interface is not None: | ||
| interface_name = self.interface.name | ||
| self.handle_error(f"{interface_name}: Timeout waiting for response") | ||
| break |
There was a problem hiding this comment.
Good bug fix for infinite loop when a command response times out and raise_exceptions=False
| BridgeInterfaceModel(name=self.name, scope=self._scope, public_key=public_key).create(force=True) | ||
| self._endpoint = await iroh.Endpoint.bind( | ||
| iroh.EndpointOptions(preset=iroh.preset_n0(), secret_key=bytes.fromhex(self._secret_key_hex)) | ||
| ) |
There was a problem hiding this comment.
When OPENC3_BRIDGE_RELAY is set to a custom or regional relay, this endpoint still uses preset_n0(), unlike the hub and host runtime. Iroh only uses relays configured on the endpoint, so remote bridges using that relay can fail. Mirror the relay-mode branch used by the other endpoints; see the Iroh relay configuration guidance: https://docs.iroh.computer/about/faq
See the implementation in host_interface_microservice.py 235-243
3 participants
Stacked PR 3 of 5 — base:
pr2-bridge-models(#3695). Part of the #3688 split. Needs PR 2.The runtime "meat" of the bridge:
bridge_microservice.py— the Iroh hub (~900 lines)bridge_interface.py— the COSMOS-side interfacehost_interface_microservice.py— the host runnercmd_response_protocol.pycompose.yamlUDP port range (7799–7814) +compose.override.yamlrelay docsNotes
bridge_microservice.py) vs client (bridge_interface.py+host_interface_microservice.py).🤖 Generated with Claude Code