summaryrefslogtreecommitdiffstats
path: root/packages/meshbay-node
diff options
context:
space:
mode:
authorChristophe Besson <cbesson@gmail.com>2026-09-07 16:02:10 +0200
committerChristophe Besson <cbesson@gmail.com>2026-09-07 16:02:10 +0200
commitd1f998b42137465b610667439527917a00030b4d (patch)
tree13674bc358ce585b0a2cd14ac210486bed79ff29 /packages/meshbay-node
parent8883d60d0afa2ed9dd1ef68bc21fe1b9a65a59ff (diff)
downloadmeshbay-d1f998b42137465b610667439527917a00030b4d.tar.gz
fix(mnp): give a reply an id, so it stops being routed by luck
MNP carried no correlation id. A reply named its own type and nothing else, so a client with more than one request in flight worked out which one a message answered from the message itself — and for the replies that name nothing it could not. `_dispatch` fell through to matching by arrival order, which is a guess. `_sendAndWait` had the right value all along: it keys `_pending` by `this._seqId++` and never put it on the wire. The guess fails asymmetrically, which is why it hid. The victim is not the request that was answered wrongly — it is the unrelated one that now waits out its own 30s timeout for a reply already delivered elsewhere. Live on 2026-09-06: five `music_meta_req` sat pending for over 100 seconds behind a failing MusicBrainz, and a `device_list_result` was handed to one of them. The composer is disabled while a send is in flight, so a chat message whose reply went astray the same way left the Chat tab looking frozen for thirty seconds, then unfroze on its own. The `ack` half of this was fixed on 2026-08-30 by matching on request type. That closed the instance and left the class open: a refusal has no type to match on either, and `_dispatch_message`'s catch-all answers every unforeseen failure with `{"type": "error", "detail": "Request failed"}` — 238 of this module's 240 error sends name nothing at all. `req_id` now rides on the request and comes back on the reply. On the node it is published for the whole handler in a ContextVar and stamped by `_send`: a parameter would have meant threading an argument through all 240 send sites, and asyncio copies the context into a task, so a handler that `_spawn`s its real work still answers under the right id. It is never stamped on a broadcast — those answer nothing, and the owner check in `_send` is what keeps a chat broadcast or an index push from reaching another peer looking like a reply. On the client, `_dispatch` resolves on `req_id` first and the arrival-order fallback is gone the moment a node proves it stamps (`_correlates`, armed by the handshake's own reply). The fallback stays for an MNP 1.0 node, unchanged and no wider: there it is the only thing there is, and removing it would leave device_list_result, join_result and the handshake replies reaching nobody. Two things fall out. `sendChat` refuses an `error` reply like every other request in the file — it returned it as success, which did not matter while a refusal reached the wrong caller anyway and would now show a rejected message as sent. And `_group_ctx` uses `.get`: a reload pops a removed group while sessions connected to it are open, and every request they had left raised KeyError into that same catch-all. Sealed index messages are the one exception to the fast path. They cannot be handed over until they are opened, which is asynchronous while `_dispatch` is not — resolving on the id alone gave `fetchIndex` the envelope and skipped `onIndexSync` entirely. Caught by extending `index_seal_probe.mjs` to stamp a reply the way a current node does, after the hub suite passed over it: the probe built its own frames and had never seen one. Tests, all failing before and passing after: `test_chat_send.py` drives the real ChatPanel over the real transport for both shapes of reply with an older request pending (3 of its 6 are new, and the 3 for `ack` pass either way, so it discriminates); `test_reply_correlation.py` pins the node's half — the refusals that name nothing else, the broadcast that must not be stamped, and a late reply from a spawned task answering under its own id rather than the most recent request's. Full suite: 1897 passed, same 11 pre-existing failures as before. QUIC keeps its own dispatch and is not stamped. It is disabled by default and no browser request reaches it, but the asymmetry is real. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01Dn1xYx9uT69mCB6UDvyKAN
Diffstat (limited to 'packages/meshbay-node')
-rw-r--r--packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py62
-rw-r--r--packages/meshbay-node/tests/test_reply_correlation.py147
2 files changed, 208 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:
diff --git a/packages/meshbay-node/tests/test_reply_correlation.py b/packages/meshbay-node/tests/test_reply_correlation.py
new file mode 100644
index 0000000..7c2138b
--- /dev/null
+++ b/packages/meshbay-node/tests/test_reply_correlation.py
@@ -0,0 +1,147 @@
+"""
+A reply names the request it answers.
+
+MNP carried no correlation id until 2026-09-07. A reply named its own type and
+nothing else, so a client with more than one request outstanding had to work out
+which one a message answered from the message itself — and for the replies that
+name nothing, it could not. This module sends `{"type": "error"}` from 240
+places and two of them say what they are about; `_dispatch_message`'s catch-all
+is one of the 238. Such a refusal reached no caller at all: the browser handed
+it to whichever request happened to be waiting, and the request it belonged to
+sat until its own 30s timeout. Live symptom (2026-09-06): the Chat composer is
+disabled while a send is in flight, so a chat message whose refusal went astray
+froze the tab for thirty seconds.
+
+`req_id` is the client's own pending-map key, put on the wire and stamped back
+onto the reply by `_send`. What matters here, and what the browser cannot check
+for itself:
+
+ * a reply carries it, including the refusals that name nothing else;
+ * a *broadcast* does not — it answers no request, and stamping it would hand
+ another peer's client a reply to a request it never made;
+ * work handed to a background task still answers under the right id, which is
+ why this is a ContextVar and not an attribute on the session.
+"""
+import asyncio
+
+import msgpack
+import pytest
+from meshbay_common.protocol import MNP
+from meshbay_node.transport.webrtc_server import WebRTCPeerSession
+
+pytestmark = pytest.mark.asyncio
+
+
+class _Channel:
+ readyState = "open"
+
+ def __init__(self):
+ self.sent = []
+
+ def send(self, framed):
+ # Skip the 4-byte length prefix _pack writes.
+ self.sent.append(msgpack.unpackb(framed[4:], raw=False))
+
+
+def _session(peer_id="p"):
+ s = WebRTCPeerSession.__new__(WebRTCPeerSession)
+ s._ctx = {}
+ s._peer_id = peer_id
+ s._user_id = None
+ s._group_id = ""
+ s._channel = _Channel()
+ s._tasks = set()
+ return s
+
+
+async def test_a_refusal_that_names_nothing_else_names_the_request():
+ """The reply at the root of the defect: no type of its own to match on."""
+ s = _session()
+ # No handshake yet, so any other message is refused — a bare `error`, the
+ # same shape the catch-all sends and the same shape a browser could not
+ # route.
+ s._handle_message({"type": MNP.CHAT_MESSAGE, "req_id": 41})
+
+ (reply,) = s._channel.sent
+ assert reply["type"] == "error"
+ assert reply["req_id"] == 41, (
+ "a refusal that names neither the request nor a type of its own is a "
+ "reply no caller can claim")
+
+
+async def test_the_catch_all_refusal_names_the_request_too():
+ """Every failure in the dispatch loop funnels into one generic reply."""
+ s = _session()
+ s._user_id = "u"
+
+ def _boom(msg):
+ raise RuntimeError("filesystem path that must not reach the peer")
+ s._do_chat_message = _boom
+
+ s._handle_message({"type": MNP.CHAT_MESSAGE, "req_id": 7})
+
+ (reply,) = s._channel.sent
+ assert reply == {"type": "error", "detail": "Request failed", "req_id": 7}, (
+ "the catch-all is where an unforeseen failure ends up, so it is exactly "
+ "the reply that must still be routable")
+
+
+async def test_a_request_without_an_id_is_answered_without_one():
+ """An older client sends none; nothing may be invented for it."""
+ s = _session()
+ s._handle_message({"type": MNP.CHAT_MESSAGE})
+
+ (reply,) = s._channel.sent
+ assert "req_id" not in reply
+
+
+async def test_a_broadcast_to_another_peer_is_not_stamped():
+ """The reply goes to the asker; the broadcast goes to everyone else.
+
+ They travel out of the same handler, and only the first answers anything.
+ Stamping the second would hand another browser a reply keyed to a pending
+ request of its own that it never sent — the very confusion this fixes.
+ """
+ asker, other = _session("asker"), _session("other")
+ asker._user_id, other._user_id = "a", "b"
+ registry = {"a": asker, "b": other}
+ asker._peer_registry = lambda: registry
+ asker._user_names = lambda: {}
+ asker._audit = lambda *a, **k: None
+ asker._group_ctx = lambda: {}
+
+ asker._handle_message({"type": MNP.CHAT_MESSAGE, "payload": "hi", "req_id": 3})
+
+ (ack,) = asker._channel.sent
+ assert ack["type"] == "ack" and ack["req_id"] == 3
+ (broadcast,) = other._channel.sent
+ assert broadcast["type"] == MNP.CHAT_MESSAGE
+ assert "req_id" not in broadcast, (
+ "a broadcast answers no request and must not look like a reply")
+
+
+async def test_work_handed_to_a_task_still_answers_under_the_right_id():
+ """Most handlers `_spawn` their real work, and the reply leaves long after
+ the dispatch call that started it has returned.
+
+ This is the reason the id lives in a ContextVar: asyncio copies the current
+ context into a task, so the answer keeps the id even though nothing passed
+ it along. An attribute on the session would have been overwritten by the
+ next message to arrive in the meantime.
+ """
+ s = _session()
+ s._user_id = "u"
+
+ async def _late(reply):
+ await asyncio.sleep(0.01)
+ s._send({"type": "roster_read_resp", "detail": reply})
+ s._do_chat_message = lambda msg: s._spawn(_late(msg["payload"]))
+
+ s._handle_message({"type": MNP.CHAT_MESSAGE, "payload": "first", "req_id": 11})
+ # A second request arrives while the first one's task is still asleep.
+ s._handle_message({"type": MNP.CHAT_MESSAGE, "payload": "second", "req_id": 12})
+ await asyncio.gather(*list(s._tasks))
+
+ by_id = {m["req_id"]: m["detail"] for m in s._channel.sent}
+ assert by_id == {11: "first", 12: "second"}, (
+ "a late reply answered under whichever request arrived most recently")