diff options
Diffstat (limited to 'packages/meshbay-node/src/meshbay_node')
| -rw-r--r-- | packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py | 62 |
1 files changed, 61 insertions, 1 deletions
diff --git a/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py b/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py index 8e357c9..aaf3f81 100644 --- a/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py +++ b/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py @@ -24,6 +24,7 @@ Signaling flow (handled externally by the hub): import asyncio import base64 +import contextvars import hashlib import hmac import logging @@ -258,6 +259,32 @@ def _pack(obj: dict) -> bytes: _WEBRTC_TRACE = os.environ.get("MESHBAY_WEBRTC_TRACE") == "1" _WEBRTC_TRACE_INTERVAL_S = 30.0 +# The request this session is currently answering, as (session, req_id). +# +# MNP has never carried a correlation id: a reply named its own type and +# nothing else, so a client with more than one request outstanding had to guess +# which one a message answered — by arrival order, for every reply the client +# could not key off a field of its own. The guess is wrong whenever two replies +# reorder, and catastrophically wrong for the replies that name *nothing*: this +# module sends `{"type": "error"}` from 240 places and two of them name what +# they are about. A refusal therefore reached no caller at all, and the request +# it belonged to waited out the client's 30s timeout while some unrelated +# request was resolved with the refusal instead. Live symptom, found 2026-09-06: +# the Chat composer is disabled while a send is in flight, so a chat message +# whose reply went astray froze the tab for 30 seconds. +# +# `req_id` closes it: whatever the caller put on the request is stamped on the +# reply. A ContextVar rather than a parameter because the alternative is +# threading an argument through all 240 send sites — and asyncio copies the +# current context into a task, so a handler that `_spawn`s its real work still +# answers under the id of the request that started it. +# +# The session is held alongside the id because a handler may send to *other* +# sessions as well as its own (a chat broadcast, an index push): those are not +# replies to anything and must not be stamped. _send checks the owner. +_REPLY_TO: contextvars.ContextVar[tuple] = contextvars.ContextVar( + "meshbay_reply_to", default=(None, None)) + class _DataChannelBuffer: """ @@ -393,6 +420,22 @@ class WebRTCPeerSession: ) def _handle_message(self, msg: dict) -> None: + """Answer one MNP message, under the correlation id it carries. + + The id is published for the whole handler — see _REPLY_TO — so that + every reply _send puts on the wire, including the ones a spawned task + sends much later and the generic refusal below, names the request it + answers. Resetting on the way out only clears it for *this* call: a + task spawned in between captured its own copy of the context when it + was created and keeps answering under the right id. + """ + token = _REPLY_TO.set((self, msg.get("req_id"))) + try: + self._dispatch_message(msg) + finally: + _REPLY_TO.reset(token) + + def _dispatch_message(self, msg: dict) -> None: mtype = msg.get("type") log.debug("WebRTC recv: %s", mtype) try: @@ -2941,7 +2984,14 @@ class WebRTCPeerSession: def _group_ctx(self) -> dict: if "groups" in self._ctx and self._group_id: - return self._ctx["groups"][self._group_id] + # `.get`, not a bare subscript. A config reload removes a group + # from this map (daemon.py's reload does `groups_ctx.pop`) while + # sessions connected to it are still open, and the next request + # any of them made raised KeyError into _dispatch_message's + # catch-all. An absent group now reads the way an unconfigured + # one already does — the handlers all test for what they need — + # instead of failing every request the session has left. + return self._ctx["groups"].get(self._group_id) or {} return self._ctx def _indexing_status(self) -> dict: @@ -4918,6 +4968,16 @@ class WebRTCPeerSession: self._audit("stream_video", entry.name) def _send(self, obj: dict) -> None: + # Stamp the reply with the id of the request being answered, so the + # caller never has to guess. Only for this session's own replies: a + # handler that also pushes to other peers (a chat broadcast, an index + # delta) reaches them through *their* _send, where the owner no longer + # matches and nothing is stamped — those messages answer no request. + # An explicit req_id already on the object wins, and an unsolicited + # push (no request in scope) carries none, exactly as before. + owner, req_id = _REPLY_TO.get() + if req_id is not None and owner is self and "req_id" not in obj: + obj = {**obj, "req_id": req_id} if self._channel and self._channel.readyState == "open": self._channel.send(_pack(obj)) else: |