From d1f998b42137465b610667439527917a00030b4d Mon Sep 17 00:00:00 2001 From: Christophe Besson Date: Mon, 7 Sep 2026 16:02:10 +0200 Subject: fix(mnp): give a reply an id, so it stops being routed by luck MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 Claude-Session: https://claude.ai/code/session_01Dn1xYx9uT69mCB6UDvyKAN --- CLAUDE.md | 27 +++- .../meshbay-common/src/meshbay_common/protocol.py | 22 +++ .../src/meshbay_hub/static/transport.js | 99 ++++++++++++-- .../meshbay-hub/tests/harness/chat_send_probe.py | 147 +++++++++++++-------- packages/meshbay-hub/tests/test_chat_send.py | 100 +++++++++----- .../meshbay-hub/tests/test_index_seal_client.py | 30 ++++- .../src/meshbay_node/transport/webrtc_server.py | 62 ++++++++- .../meshbay-node/tests/test_reply_correlation.py | 147 +++++++++++++++++++++ 8 files changed, 525 insertions(+), 109 deletions(-) create mode 100644 packages/meshbay-node/tests/test_reply_correlation.py diff --git a/CLAUDE.md b/CLAUDE.md index 4a4175f..04f6e61 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -471,11 +471,28 @@ anything that assumes one key per person. next visit to the tab, since the node had stored it and answered. Every line of `chat-app.js` is correct and every routed message in `transport.js` is routed correctly; the defect is in the seam, which is why - `tests/harness/chat_send_probe.py` drives the two together. `ack` is now - matched by request type (`chat_msg`, or the keypair-bundle store/delete that - name themselves in `detail`). Anything left to the "oldest pending" guess is - a latent version of this bug: a request type deserves a key, and a reply - deserves something to key it by + `tests/harness/chat_send_probe.py` drives the two together. `ack` was matched + by request type (`chat_msg`, or the keypair-bundle store/delete that name + themselves in `detail`), and that closed the instance — **but it left the + class open, and it came back on 2026-09-06 through the other door.** A + refusal has no type of its own to key on: `_dispatch_message`'s catch-all + answers every unforeseen failure with `{"type": "error", "detail": "Request + failed"}`, and 238 of `webrtc_server.py`'s 240 error sends name nothing + either. So a chat send the node refused was routed by luck all over again — + same frozen composer, same 30 s, and rare enough (it needs an older request + still waiting, which a `music_meta_req` behind a failing third-party lookup + supplies for over a hundred seconds) to look like once every couple of days. + The keys were never the fix, only a workaround for a protocol that carried no + correlation id at all: `_seqId` existed, indexed `_pending`, and was never put + on the wire. It is now (`req_id`, see `protocol.py`) — the node stamps it on + the reply from `_send`, via a ContextVar so a handler's spawned work still + answers under the right id, and never on a broadcast, which answers nothing. + With that, the arrival-order fallback is gone for any node that stamps. + The lesson is not "key the replies": it is that **matching by arrival order + is a guess that fails silently and asymmetrically** — the victim is never + the request that was answered wrongly, it is the unrelated one that now + waits for a reply already delivered elsewhere. A reply needs an identifier + the protocol guarantees, not a field it happens to have - **A refusal that never rejects.** Denying Chromium's `fullscreen` permission does not make `requestFullscreen()` throw — the promise never settles. The diff --git a/packages/meshbay-common/src/meshbay_common/protocol.py b/packages/meshbay-common/src/meshbay_common/protocol.py index 689adbf..6eff5b6 100644 --- a/packages/meshbay-common/src/meshbay_common/protocol.py +++ b/packages/meshbay-common/src/meshbay_common/protocol.py @@ -6,6 +6,28 @@ MHP (Mesh Bay Hub Protocol) — v0.1 All wire messages are length-prefixed msgpack (4-byte big-endian length header). Every message carries a "v" field for protocol version. + +**`req_id` — the correlation id (added 2026-09-07).** A request may carry one; +the reply to it carries the same value back, and nothing else on the wire does. +It is the caller's own key for its pending request, opaque to the node, and +unique only within one connection. + +There was none for a long time, and its absence was not neutral. A reply named +its own type and nothing else, so a caller with more than one request in flight +had to work out which one a message answered from the message itself — and the +replies that name nothing (a bare `ack`, and `{"type": "error"}`, which +webrtc_server.py sends from 240 places while two of them say what they are +about) could only be matched by arrival order. That is a guess, wrong whenever +two replies reorder, and it does not fail quietly: one request is resolved with +another's answer while the request that answer belonged to waits out its own +timeout. Live symptom (2026-09-06): a chat send whose reply went astray left +the composer disabled for thirty seconds, and the Chat tab read as frozen. + +Both halves are optional and degrade to what came before: a request without one +is answered without one, and a client that gets no id back falls back to +matching by type. Neither side may treat it as authentication or as a sequence +number — it is a label chosen by the peer, and the only thing it decides is +which local promise a reply belongs to. """ from dataclasses import dataclass, field diff --git a/packages/meshbay-hub/src/meshbay_hub/static/transport.js b/packages/meshbay-hub/src/meshbay_hub/static/transport.js index 179292e..e57ad8b 100644 --- a/packages/meshbay-hub/src/meshbay_hub/static/transport.js +++ b/packages/meshbay-hub/src/meshbay_hub/static/transport.js @@ -287,6 +287,11 @@ class MeshBayTransport { this._pc = null; this._channel = null; this._pending = new Map(); + // Set the first time this connection sees a reply that names the request + // it answers (see _dispatch). A node either stamps every reply or none, + // so one is proof for the connection — and once there is proof, the + // arrival-order fallback at the bottom of _dispatch is never right again. + this._correlates = false; this._seqId = 0; this._recvBuf = new Uint8Array(0); this._connected = false; @@ -1429,6 +1434,12 @@ class MeshBayTransport { thread_id: threadId || null, sender_name: senderName || null, }); + // Every other request in this file refuses an `error` reply; this one + // returned it as though the node had accepted the message. It never + // mattered while a refusal reached the wrong caller anyway — now that a + // reply finds the request that made it, a message the node rejected + // would otherwise appear in the conversation as sent. + if (msg.type === 'error') throw new Error(msg.detail || 'chat send refused'); return msg; } @@ -2191,12 +2202,15 @@ class MeshBayTransport { if (msg.type === 'index_sync') { if (this._onIndexSync) this._onIndexSync(opened); - for (const [, handler] of this._pending) { - if (handler._reqType === 'index_sync') { - handler.resolve(opened); - break; - } - } + // The node's first push to a newly connected peer is an index_sync + // nobody asked for, so there is not always a request to resolve. When + // there is, `req_id` says which one — the type match below is what a + // node too old to stamp one leaves us, and it is why two fetches in + // flight at once used to resolve the wrong one. + const handler = opened.req_id !== undefined && opened.req_id !== null + ? this._pending.get(opened.req_id) + : [...this._pending.values()].find(h => h._reqType === 'index_sync'); + if (handler) handler.resolve(opened); return; } if (this._onIndexDelta) this._onIndexDelta(opened); @@ -2283,7 +2297,13 @@ class MeshBayTransport { resolve: (msg) => { clearTimeout(timeout); this._pending.delete(id); resolve(msg); }, reject: (err) => { clearTimeout(timeout); this._pending.delete(id); reject(err); }, }); - this._send(obj); + // The id goes on the wire (MNP 1.1+): a node that understands it stamps + // the reply with it, and _dispatch matches on that alone. It used to be + // local to this map, which is why every reply had to be recognised by + // some field of its own — and why the ones that carry no such field + // reached their caller by luck. An older node ignores the extra key and + // is routed by the per-type fallbacks below, exactly as before. + this._send({ ...obj, req_id: id }); }); } @@ -2323,6 +2343,45 @@ class MeshBayTransport { } _dispatch(msg) { + // A reply that names the request it answers. Nothing below this needs to + // recognise it, and nothing below this may see it: every remaining branch + // exists to identify a reply by some field of its own, which is the job + // this makes unnecessary. + // + // What is left underneath is genuinely unsolicited — a broadcast to every + // connected client, a push, a challenge — or a reply from a node too old + // to stamp one, which is what the per-type keys are for now. + if (msg.req_id !== undefined && msg.req_id !== null) { + this._correlates = true; + // The one exception, and the only one: an index message is sealed under + // the GEK and cannot be handed to its caller until it is opened, which + // is not something this synchronous function can do. Resolving it here + // would give `fetchIndex` the envelope — nonce and ciphertext, no + // entries — and skip `_onIndexSync` entirely. `_queueIndexMessage` + // opens it and then resolves, by this same id. + const sealed = msg.type === 'index_sync' || msg.type === 'index_delta'; + if (!sealed) { + const handler = this._pending.get(msg.req_id); + if (handler) { + handler.resolve(msg); + // The acks whose *broadcast* half their own requester also needs: + // every other client learns the change from the broadcast, and the + // one that asked for it is the only one that would not, because its + // own request swallowed its copy. Same call the keyed `_ack` branch + // below makes, for the same reason. + if (BROADCAST_ACK_TYPES.has(msg.type)) _replayBroadcast(this, msg); + return; + } + // Answers a request that is no longer waiting: it gave up at its own + // timeout, or a reconnect rejected everything in flight. It belongs to + // nobody, and the whole point of this change is that it is not offered + // to somebody else instead. + console.warn('[MeshBay] late reply to req', msg.req_id, '(', msg.type, + ') — nothing waiting'); + return; + } + } + // Two-step admin-op flow (_authorizeAdminOp, ADMIN_OP_TYPES) — resolve // by (op) key before anything below gets a chance to steal it via the // generic "oldest pending" fallback further down. Returns as soon as a @@ -2715,10 +2774,28 @@ class MeshBayTransport { return; } - // Everything above is routed by something in the message. What is left is - // matched by arrival order, which is only ever a guess — and a wrong guess - // here hands one request's answer to another, which then waits for a reply - // that already came. Logged so that guess is visible. + // Everything above is routed by something in the message. What is left + // used to be matched by arrival order — a guess, and a wrong guess hands + // one request's answer to another, which then waits out its own 30s + // timeout for a reply that already came and went. That is how the Chat + // composer, disabled while a send is in flight, could stay disabled for + // thirty seconds on a message the node had already stored. + // + // A node that stamps its replies (`req_id`, handled at the top) has taken + // every one of its answers out of this path, so anything arriving here is + // unsolicited and the guess can only ever be wrong. Dropping it loses + // nothing and stops the theft. + if (this._correlates) { + console.warn('[MeshBay] unsolicited', msg.type, '— dropped (pending:', + this._pending.size, ')'); + return; + } + + // Only a node too old to stamp anything reaches here, where arrival order + // is still the only thing there is. Kept deliberately, and no wider than + // it was: the alternative for such a node is that half the protocol + // (device_list_result, join_result, the handshake's own replies) reaches + // nobody at all. const oldest = this._pending.entries().next(); if (!oldest.done) { const [, handler] = oldest.value; diff --git a/packages/meshbay-hub/tests/harness/chat_send_probe.py b/packages/meshbay-hub/tests/harness/chat_send_probe.py index f1cc191..98983f3 100644 --- a/packages/meshbay-hub/tests/harness/chat_send_probe.py +++ b/packages/meshbay-hub/tests/harness/chat_send_probe.py @@ -20,8 +20,25 @@ never appeared, while the node had stored it all along. chat_send_probe.py -Prints JSON: `steps`, the state of the panel at each stage, and `log`, what the -transport sent and how the deliberately-unanswered request ended up. +It now drives two shapes of reply, because there were two ways for one to go +astray and only the first was ever fixed: + + `ack` the node accepts the message and answers `{"type": "ack"}`, which + names no request. Routed by request type since 2026-08-30. + `error` the node *refuses* it — every failure in `_dispatch_message` ends at + one catch-all sending `{"type": "error", "detail": "Request failed"}`, + and 238 of this module's 240 error sends name nothing either. That + reply reached no caller at all: it went to whatever request happened + to be waiting, and the send sat out its own 30s timeout with the + composer disabled. + +Both are run with an older request already pending — the condition that turns +"guess by arrival order" from usually-right into wrong — and both must come +back inside a second and a half. + +Prints JSON: `scenarios`, the state of the panel at each stage of each, and +`log`, what the transport sent and how the deliberately-unanswered request +ended up. """ import http.server import json @@ -54,13 +71,6 @@ import { ChatPanel } from '/chat-app.js'; const log = []; window.addEventListener('error', e => log.push('error: ' + e.message)); -// The real transport, with only the channel replaced: _send takes the plain -// object _sendAndWait built, so the framing and msgpack are the only things -// skipped — every pending entry, key and dispatch path below is the shipped one. -const tp = new window.MeshBayTransport('', 'token'); -tp._connected = true; -tp._channel = { readyState: 'open', send() {} }; - const now = Date.now() / 1000; const history = []; for (let i = 0; i < 5; i++) { @@ -68,50 +78,45 @@ for (let i = 0; i < 5; i++) { payload: 'message ' + i, timestamp: now - (5 - i) * 60 }); } -// Stands in for the node, answering exactly what webrtc_server.py answers. -// media_meta_req is answered with nothing at all, which is what a refusal -// amounts to for the request that asked: `_do_media_meta_request` sends a bare -// `error` for a file_id the index does not have, and a bare error names no -// request, so it reaches none. -tp._send = (obj) => { - log.push('sent ' + obj.type); - if (obj.type === 'chat_hist') { - setTimeout(() => tp._dispatch( - { type: 'chat_hist_resp', v: '0.2', messages: history, has_more: false }), 10); - } else if (obj.type === 'chat_msg') { - setTimeout(() => tp._dispatch({ type: 'ack', v: '0.14' }), 10); - } -}; - -function Host() { +const wait = ms => new Promise(r => setTimeout(r, ms)); + +// Stands in for the node, answering what webrtc_server.py answers — including +// stamping the reply with the id of the request it is answering, which is what +// `_send` does there. `chatReply` is the only difference between the two runs. +function makeTransport(chatReply) { + // The real transport, with only the channel replaced: _send takes the plain + // object _sendAndWait built, so the framing and msgpack are the only things + // skipped — every pending entry, key and dispatch path below is the shipped + // one. + const tp = new window.MeshBayTransport('', 'token'); + tp._connected = true; + tp._channel = { readyState: 'open', send() {} }; + tp._send = (obj) => { + log.push('sent ' + obj.type); + const answer = (reply) => setTimeout( + () => tp._dispatch({ ...reply, req_id: obj.req_id }), 10); + if (obj.type === 'chat_hist') { + answer({ type: 'chat_hist_resp', v: '0.2', messages: history, has_more: false }); + } else if (obj.type === 'chat_msg') { + answer(chatReply); + } + // music_meta_req is answered by nothing at all, on purpose: the node holds + // one open for as long as the third-party lookup behind it takes, which + // was measured at over 100 seconds with that service failing. It is the + // older pending request every scenario here needs. + }; + return tp; +} + +function Host({ tp }) { const transportRef = useRef(tp); const gekRef = useRef(null); return html`<${ChatPanel} transportRef=${transportRef} gekRef=${gekRef} username="me" entries=${[]} status="connected" />`; } -render(html`<${Host} />`, document.getElementById('root')); - -const out = { steps: [], log }; -const composer = () => document.querySelector('.chat-input'); - -function snap(label) { - const c = composer(); - out.steps.push({ - label, - bubbles: document.querySelectorAll('.chat-bubble').length, - lastText: [...document.querySelectorAll('.chat-text')].pop()?.textContent ?? null, - // What a frozen tab actually is: the composer is disabled for as long as - // a send is in flight. - composerDisabled: c ? c.disabled : null, - composerValue: c ? c.value : null, - pending: tp._pending.size, - }); -} -const wait = ms => new Promise(r => setTimeout(r, ms)); - -function typeInto(text) { - const c = composer(); +function typeInto(root, text) { + const c = root.querySelector('.chat-input'); c.focus(); // Preact reads e.target.value on input, so the native setter has to run. Object.getOwnPropertyDescriptor(window.HTMLTextAreaElement.prototype, 'value') @@ -119,22 +124,42 @@ function typeInto(text) { c.dispatchEvent(new Event('input', { bubbles: true })); } -(async () => { +async function runScenario(name, chatReply) { + const root = document.createElement('div'); + document.getElementById('root').appendChild(root); + const tp = makeTransport(chatReply); + render(html`<${Host} tp=${tp} />`, root); + + const steps = []; + const snap = (label) => { + const c = root.querySelector('.chat-input'); + steps.push({ + label, + bubbles: root.querySelectorAll('.chat-bubble').length, + lastText: [...root.querySelectorAll('.chat-text')].pop()?.textContent ?? null, + // What a frozen tab actually is: the composer is disabled for as long as + // a send is in flight. + composerDisabled: c ? c.disabled : null, + composerValue: c ? c.value : null, + pending: tp._pending.size, + }); + }; + await wait(500); snap('arrived'); - // The Videos tab asked about a file a moment ago and is still waiting. Any - // unanswered request will do; this is the one that was live when the defect - // was found. - tp.fetchMediaMeta('a-file-the-node-refused') - .then(m => log.push('media_meta resolved with ' + m.type), - e => log.push('media_meta rejected: ' + e.message)); + // The Music tab asked about a track and is still waiting on the node, which + // is waiting on something else. Any older unanswered request will do; this + // is the one that was live when the defect was found. + tp.fetchMusicMeta('a-track-the-node-is-slow-about') + .then(m => log.push(name + ': music_meta resolved with ' + m.type), + e => log.push(name + ': music_meta rejected: ' + e.message)); await wait(100); - snap('stale request pending'); + snap('older request pending'); - typeInto('hello'); + typeInto(root, 'hello'); await wait(100); - composer().dispatchEvent(new KeyboardEvent('keydown', + root.querySelector('.chat-input').dispatchEvent(new KeyboardEvent('keydown', { key: 'Enter', bubbles: true, cancelable: true })); // Far short of _sendAndWait's 30s timeout: a send that has not come back by @@ -142,6 +167,14 @@ function typeInto(text) { await wait(1500); snap('after send'); + return { name, steps }; +} + +(async () => { + const out = { scenarios: [], log }; + out.scenarios.push(await runScenario('ack', { type: 'ack', v: '0.14' })); + out.scenarios.push(await runScenario( + 'error', { type: 'error', detail: 'Request failed' })); fetch('/log', { method: 'POST', body: JSON.stringify(out) }); })(); """ diff --git a/packages/meshbay-hub/tests/test_chat_send.py b/packages/meshbay-hub/tests/test_chat_send.py index 4224fdb..8b098c2 100644 --- a/packages/meshbay-hub/tests/test_chat_send.py +++ b/packages/meshbay-hub/tests/test_chat_send.py @@ -1,21 +1,27 @@ """ -Sending a chat message must come back. - -The node answers a chat message with a bare `{"type": "ack"}` — no request id, -no type of its own — so `_dispatch` had nothing to match it on and left it to -the arrival-order guess at the end of the function. That guess is wrong as soon -as anything else this browser asked for is still waiting: the ack was handed to -*that* request, and the send waited out `_sendAndWait`'s 30s timeout. Since the -composer is disabled while a send is in flight, the Chat tab stopped taking -clicks and keys, the message never appeared — and it was there on the next -visit, because the node had stored it and answered. - -An outstanding request is the ordinary case, not a rare one: the node refuses -an unknown file_id with a bare `error`, which names no request either, so a -Videos tab that asked about a file the index no longer has leaves a -`media_meta_req` in `_pending` for a full 30s. - -None of that is visible in `chat-app.js`, where every line is correct, so this +Sending a chat message must come back — accepted or refused. + +The node's replies to a chat message name no request. The acceptance is a bare +`{"type": "ack"}`; the refusal is a bare `{"type": "error"}`, and it is not a +special case — `_dispatch_message`'s catch-all answers *every* failure that +way, and 238 of webrtc_server.py's 240 error sends name nothing either. So +`_dispatch` had nothing to match either reply on and left both to the +arrival-order guess at the end of the function. + +That guess is wrong as soon as anything else this browser asked for is still +waiting, which is the ordinary case rather than a rare one: a `music_meta_req` +sits in `_pending` for as long as the third-party lookup behind it takes, and +that was measured live at over 100 seconds with the service failing. The reply +went to *that* request, and the send waited out `_sendAndWait`'s 30s timeout. +Since the composer is disabled while a send is in flight, the Chat tab stopped +taking clicks and keys, and the message never appeared. + +The ack half was fixed by matching on request type. The refusal half could not +be: an `error` has no type of its own to match on. `req_id` is what closed it — +the caller's id, stamped on the reply by the node — so this now drives both +shapes of answer. + +None of it is visible in `chat-app.js`, where every line is correct, so this drives the real panel over the real transport in a browser rather than reading either source. """ @@ -39,35 +45,61 @@ def probe(): run = subprocess.run(["python3", str(HARNESS)], capture_output=True, timeout=180) assert run.returncode == 0, run.stderr.decode()[-2000:] data = json.loads(run.stdout.decode()) - return data, {s["label"]: s for s in data["steps"]} + steps = {sc["name"]: {s["label"]: s for s in sc["steps"]} + for sc in data["scenarios"]} + return data, steps + +@pytest.mark.parametrize("reply", ["ack", "error"]) +def test_the_composer_comes_back(probe, reply): + """The one thing a person sees: the tab is usable again. -def test_the_composer_comes_back(probe): - """The one thing a person sees: the tab is usable again.""" + Both answers have to release it. A refusal that reaches nobody leaves the + composer disabled exactly as long as an acceptance that reaches nobody — + the composer is not waiting for good news, it is waiting for an answer. + """ _, steps = probe - assert steps["stale request pending"]["composerDisabled"] is False, ( + assert steps[reply]["older request pending"]["composerDisabled"] is False, ( "the composer was already unusable before the send") - assert steps["after send"]["composerDisabled"] is False, ( + assert steps[reply]["after send"]["composerDisabled"] is False, ( "the composer is still disabled well inside the 30s request timeout -- " "the send never came back, which is what reads as a frozen Chat tab") -def test_the_message_is_displayed(probe): +def test_an_accepted_message_is_displayed(probe): """A sent message appears at once, not on the next visit to the tab.""" _, steps = probe - before = steps["stale request pending"]["bubbles"] - assert steps["after send"]["bubbles"] == before + 1, ( + before = steps["ack"]["older request pending"]["bubbles"] + assert steps["ack"]["after send"]["bubbles"] == before + 1, ( "the message was not added to the conversation") - assert steps["after send"]["lastText"] == "hello" - assert steps["after send"]["composerValue"] == "", ( + assert steps["ack"]["after send"]["lastText"] == "hello" + assert steps["ack"]["after send"]["composerValue"] == "", ( "the text came back into the composer, so the send was treated as failed") -def test_the_ack_is_not_handed_to_another_request(probe): - """The other half of the same defect: whatever was waiting got the ack and - carried on with a reply to a question it never asked.""" +def test_a_refused_message_is_not_displayed_as_sent(probe): + """The other direction, and the one routing this correctly makes possible. + + While a refusal reached the wrong caller it did not matter what `sendChat` + would have done with it. Now that it arrives, a message the node rejected + must not appear in the conversation as though it had been stored — it must + come back into the composer, where a person can see it did not go. + """ + _, steps = probe + before = steps["error"]["older request pending"]["bubbles"] + assert steps["error"]["after send"]["bubbles"] == before, ( + "a refused message was added to the conversation anyway") + assert steps["error"]["after send"]["composerValue"] == "hello", ( + "the refused text was dropped instead of being handed back") + + +@pytest.mark.parametrize("reply", ["ack", "error"]) +def test_the_reply_is_not_handed_to_another_request(probe, reply): + """The other half of the same defect: whatever was waiting got the reply + and carried on with an answer to a question it never asked.""" data, _ = probe - assert "media_meta resolved with ack" not in data["log"], ( - "the chat ack was routed to the pending media_meta_req -- that request " - "now believes it has an answer, and the chat send is waiting for a " - "reply that already arrived") + stolen = [line for line in data["log"] if line.startswith(f"{reply}: music_meta")] + assert not stolen, ( + f"the chat {reply} was routed to the pending music_meta_req ({stolen}) -- " + "that request now believes it has an answer, and the chat send is " + "waiting for a reply that already arrived") diff --git a/packages/meshbay-hub/tests/test_index_seal_client.py b/packages/meshbay-hub/tests/test_index_seal_client.py index ca2c7a2..e1135da 100644 --- a/packages/meshbay-hub/tests/test_index_seal_client.py +++ b/packages/meshbay-hub/tests/test_index_seal_client.py @@ -45,10 +45,15 @@ def _frame(msg: dict) -> str: return (struct.pack(">I", len(body)) + body).hex() -def _sync_frame(gek: bytes, *names: str, version: int = 3) -> str: +def _sync_frame(gek: bytes, *names: str, version: int = 3, + req_id: int | None = None) -> str: payload = {"version": version, "entries": [_entry(n) for n in names], "dirs": ["library"], "roots": [{"name": "library"}]} + # `req_id` is what a current node stamps on a *reply*; the push it sends a + # newly connected peer answers no request and carries none. Both shapes + # arrive here, and only one of them may resolve a waiting fetchIndex. return _frame({"type": "index_sync", "v": "1.0", "group_id": GROUP, + **({"req_id": req_id} if req_id is not None else {}), **seal(gek, PURPOSE_INDEX, "index_sync", GROUP, payload)}) @@ -141,3 +146,26 @@ def test_deltas_are_applied_in_arrival_order(): assert [e["additions"][0] for e in out["events"][1:]] == [ "added-0.mkv", "added-1.mkv", "added-2.mkv"] assert [e["base_version"] for e in out["events"][1:]] == [3, 4, 5] + + +def test_a_sealed_reply_is_opened_before_it_reaches_its_caller(): + """A stamped index_sync must not be short-circuited by its `req_id`. + + Every other reply a node stamps is resolved straight out of the pending + map, which is the whole point of the id. An index message cannot be: it is + sealed, opening it is asynchronous, and `_dispatch` is not. Handing it over + on the strength of the id alone gives `fetchIndex` the envelope — nonce and + ciphertext, no entries — and never calls `onIndexSync` at all. + + `req_id` is 0 here because it is the transport's first request, and a + falsy id is exactly the one a presence check gets wrong. + """ + out = _run([_sync_frame(GEK, "a-film.mkv", req_id=0)]) + + assert [e["event"] for e in out["events"]] == ["index_sync"], ( + "the consumer was never told about an index that arrived as a reply") + assert out["events"][0]["entries"] == ["a-film.mkv"] + assert out["events"][0]["hasCiphertext"] is False + assert out["fetchIndex"]["state"] == "resolved" + assert out["fetchIndex"]["entries"] == ["a-film.mkv"], ( + "the caller was handed the sealed envelope instead of the index") 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") -- cgit v1.2.3