diff options
Diffstat (limited to 'packages')
4 files changed, 375 insertions, 9 deletions
diff --git a/packages/meshbay-hub/src/meshbay_hub/static/file-utils.js b/packages/meshbay-hub/src/meshbay_hub/static/file-utils.js index 585558f..ae6cd6f 100644 --- a/packages/meshbay-hub/src/meshbay_hub/static/file-utils.js +++ b/packages/meshbay-hub/src/meshbay_hub/static/file-utils.js @@ -349,9 +349,13 @@ async function _writeOrStall(writable, bytes, at) { } } +// "Transport closed" is not here: that is close(), somebody ending the +// connection on purpose. "DataChannel closed" is the channel going away under +// it, which is what a reconnect is for. function _isRetryableTransportError(err) { return err.name === 'TransportLostError' || err.message === 'Response timeout' + || err.message === 'DataChannel closed' || (err.message || '').startsWith('DataChannel not open'); } diff --git a/packages/meshbay-hub/src/meshbay_hub/static/transport.js b/packages/meshbay-hub/src/meshbay_hub/static/transport.js index a07f431..15c7b09 100644 --- a/packages/meshbay-hub/src/meshbay_hub/static/transport.js +++ b/packages/meshbay-hub/src/meshbay_hub/static/transport.js @@ -447,6 +447,16 @@ const LEASE_WATCHDOG_MS = 60000; // one sender: 1 connection 5 to 6 MB/s, 4 about 19, 8 between 27 and 34, and // 16 no more than that (the node's sender is one Python process). const DOWNLOAD_LANES = 8; + +// What a connection sends while it is opening its session, before the node +// knows who is on it: the node answers anything else with "Handshake +// required" (webrtc/dispatch.py), and the same list is what it accepts there. +// While a reconnect runs these are the only messages that go out; everything +// else waits for the session (see `_sendAndWait`). +const SESSION_OPENING = new Set([ + 'handshake', 'handshake_response', 'keypair_bundle_fetch', 'gek_bundle_fetch', + 'join_request', 'device_add_request', 'device_hello', +]); // A lane left unused this long is closed: each one is a peer connection the // node holds open. const LANE_IDLE_MS = 30000; @@ -594,6 +604,10 @@ class MeshBayTransport { this._lanesTimer = 0; this._laneRetryAt = 0; this._lane = false; + // True from the end of connect() — the node has authenticated this side — + // until the connection is lost. Not `_connected`, which is true from the + // moment the channel opens, a whole handshake earlier. + this._sessionUp = false; // Set once close() runs — stops the automatic reconnect from firing on a // connection the caller tore down on purpose (leaving the group, page // unload), which would otherwise race back in right as everything else @@ -810,6 +824,7 @@ class MeshBayTransport { // reconnect that never got as far as announcing a device still looked, to // the composer, exactly like one that had. this._setDevicePk('', 'new connection'); + this._sessionUp = false; this._pc = new RTCPeerConnection({ iceServers: await iceServers() }); this._channel = this._pc.createDataChannel('mnp', { ordered: true }); @@ -836,15 +851,23 @@ class MeshBayTransport { channelReady.catch(() => { /* the awaiter below reports it */ }); this._channel.onmessage = (event) => this._onMessage(event.data); + const channel = this._channel; this._channel.onclose = (ev) => { - console.warn('[MeshBay] DataChannel closed', this._channel?.readyState, ev); + console.warn('[MeshBay] DataChannel closed', channel.readyState, ev); trace('channel_close', { - readyState: this._channel?.readyState, + readyState: channel.readyState, pc: this._pc?.connectionState, ice: this._pc?.iceConnectionState, }); - this._connected = false; if (channelReject) channelReject(new Error('DataChannel closed')); + // A reconnect closes the old channel before it opens the next, and the + // event comes when it comes: from a channel that is no longer this + // connection's, it is news about nothing. Acted on, it marked the new + // session closed and failed whatever it had in flight, its own + // handshake included. + if (channel !== this._channel) return; + this._connected = false; + this._sessionUp = false; for (const [, p] of this._pending) p.reject(new Error('DataChannel closed')); this._pending.clear(); }; @@ -876,6 +899,7 @@ class MeshBayTransport { // sit out its own timeout instead of failing fast. if (pc.connectionState === 'failed' && pc === this._pc && !this._closed) { this._connected = false; + this._sessionUp = false; this._reconnect(); const err = new Error('WebRTC connection lost'); err.name = 'TransportLostError'; @@ -1303,6 +1327,7 @@ class MeshBayTransport { console.warn('[MeshBay] device_hello failed — chat will not work:', e); }); + this._sessionUp = true; return ack; } @@ -1757,6 +1782,7 @@ class MeshBayTransport { if (this._channel) this._channel.close(); if (this._pc) this._pc.close(); this._connected = false; + this._sessionUp = false; for (const [, p] of this._pending) p.reject(new Error('Transport closed')); this._pending.clear(); } @@ -1823,12 +1849,25 @@ class MeshBayTransport { async _sendAndWait(obj, timeoutMs = 30000) { // A reconnect already in flight (see _reconnectLoop) means the channel - // this would send on is the one just declared dead. `_inReconnectAttempt` - // excludes the handshake connect() itself makes while reconnecting — that - // call runs *inside* this same _reconnectPromise, which cannot resolve - // until it returns, so waiting on it here would just be waiting on - // itself for the full 6s, on every step of the handshake, every time. - if (!this._inReconnectAttempt) await this.waitForReconnect(6000); + // this would send on is the one just declared dead. The handshake that + // connect() sends while reconnecting runs *inside* that same + // _reconnectPromise, which cannot resolve until it returns, so waiting on + // it here would be waiting on itself — those messages go straight out. + // + // Only those. This used to let *every* message through while the attempt + // ran, so a download's chunk asked for again in that window went out on a + // channel that had opened and not finished its handshake: the node + // answered "Handshake required", which no retry treats as a dropped + // connection, and a download was failed by the reconnect meant to save it. + const opening = SESSION_OPENING.has(obj.type); + if (!(opening && this._inReconnectAttempt)) await this.waitForReconnect(6000); + if (!opening && !this._sessionUp) { + // No session to send it on: what the caller sees on a dropped connection, + // and what the download path waits out (`_fetchChunkResilient`). + const err = new Error('WebRTC connection lost'); + err.name = 'TransportLostError'; + throw err; + } return new Promise((resolve, reject) => { const id = this._seqId++; const timeout = setTimeout(() => { diff --git a/packages/meshbay-hub/tests/harness/reconnect_race_harness.mjs b/packages/meshbay-hub/tests/harness/reconnect_race_harness.mjs new file mode 100644 index 0000000..0f1cd97 --- /dev/null +++ b/packages/meshbay-hub/tests/harness/reconnect_race_harness.mjs @@ -0,0 +1,230 @@ +// Run the SHIPPED download over the SHIPPED reconnect, against a node that +// refuses what comes before the handshake, and see whether the download lives. +// +// transport.js is executed whole — connect(), _reconnectLoop, _sendAndWait, +// _dispatch and the framing are the shipped code — and the pipeline is lifted +// out of file-utils.js as text. What is modelled is the environment: a clock, +// a peer connection that opens its channel some time after the answer, a hub +// that answers the offer and the token, and a node that answers anything but +// the opening messages with "Handshake required" until the proof is in, as +// webrtc/dispatch.py does. Cryptography is not under test here: the checks +// that would refuse a peer are stubbed to accept. +// +// The scenario is the one that failed live: the node stops answering, the +// chunks in flight time out, and their retries come due while the reconnect's +// new channel is open and its handshake is not finished. +// +// Usage: node reconnect_race_harness.mjs <file-utils.js> <transport sources> <json> +import { readFileSync } from 'fs'; +import { delimiter } from 'path'; + +const utils = readFileSync(process.argv[2], 'utf8'); +const transportSrc = process.argv[3].split(delimiter) + .map((p) => readFileSync(p, 'utf8')).join('\n'); +const { + chunks = 400, + chunkMs = 100, + hangAt = 5000, // the node stops answering the first connection + failAfter = 30200, // ...and the peer connection reports `failed` this much later + openMs = 100, // answer → channel open + rttMs = 300, // one request → its reply, during the handshake + lateCloseMs = 0, // >0: the replaced channel's close event fires this late + runUntil = 300000, +} = JSON.parse(process.argv[4] || '{}'); + +for (const level of ['log', 'warn', 'error', 'info', 'debug']) { + console[level] = (...args) => process.stderr.write(args.join(' ') + '\n'); +} +process.on('unhandledRejection', () => {}); + +// ── The clock ──────────────────────────────────────────────────────────────── + +let now = 0; +let nextId = 1; +let timers = []; +globalThis.setTimeout = (fn, ms) => { + const t = { at: now + (ms || 0), fn, id: nextId++ }; + timers.push(t); + return t.id; +}; +globalThis.setInterval = (fn, ms) => { + const t = { at: now + ms, fn, id: nextId++, every: ms }; + timers.push(t); + return t.id; +}; +globalThis.clearTimeout = (id) => { timers = timers.filter((t) => t.id !== id); }; +globalThis.clearInterval = globalThis.clearTimeout; +Date.now = () => now; +const flush = async () => { for (let i = 0; i < 50; i++) await Promise.resolve(); }; +async function run(until) { + await flush(); + while (timers.length) { + const due = timers.reduce((a, b) => (b.at < a.at ? b : a)); + if (due.at > until) break; + if (due.every) due.at += due.every; + else timers = timers.filter((t) => t !== due); + now = Math.max(now, due.every ? due.at - due.every : due.at); + due.fn(); + await flush(); + } + now = until; +} + +// ── The page around the transport ──────────────────────────────────────────── + +globalThis.window = globalThis; +globalThis.addEventListener = () => {}; +globalThis.removeEventListener = () => {}; +globalThis.location = { hash: '' }; +globalThis.document = { + addEventListener() {}, removeEventListener() {}, visibilityState: 'visible', +}; +globalThis.localStorage = { getItem: () => null, setItem() {}, removeItem() {} }; + +const b64 = (u8) => Buffer.from(u8).toString('base64'); +window.MeshBayCrypto = { + b64encode: (u8) => b64(u8), + b64decode: (s) => new Uint8Array(Buffer.from(String(s), 'base64')), + verifyNodeSignature: async () => true, + challengeTranscript: () => new Uint8Array(1), + webrtcBinding: () => new Uint8Array(1), + handshakeProof: async () => new Uint8Array([7]), + handshakeTranscript: () => new Uint8Array(1), + constantTimeEqual: () => true, + openGroup: async () => window.__codec.msgpack_encode({ chat_epoch: 1 }), + decryptChunkBin: async (_k, _f, index) => new Uint8Array([index & 255]), +}; + +const SDP = 'v=0\r\na=fingerprint:sha-256 AA:BB\r\n'; +globalThis.fetch = async (url) => ({ + ok: true, status: 200, headers: { get: () => null }, + json: async () => (url.endsWith('/mnp-token') ? { mnp_token: 't' } : { sdp: SDP }), +}); + +// ── The node ───────────────────────────────────────────────────────────────── + +const OPENING = new Set(['handshake', 'handshake_response', 'keypair_bundle_fetch', + 'gek_bundle_fetch', 'join_request', 'device_add_request', 'device_hello']); +let refusedBeforeHandshake = 0; +let served = 0; +const pcs = []; + +class NodeSession { + constructor(channel) { this.channel = channel; this.authed = false; this.hung = false; } + reply(obj, ms) { + setTimeout(() => { + if (this.hung || this.channel.readyState !== 'open') return; + const body = window.__codec.msgpack_encode(obj); + const frame = new Uint8Array(4 + body.byteLength); + new DataView(frame.buffer).setUint32(0, body.byteLength, false); + frame.set(body, 4); + this.channel.onmessage({ data: frame.buffer }); + }, ms); + } + receive(frame) { + if (this.hung) return; + const msg = window.__codec.msgpack_decode(frame.slice(4)); + const req_id = msg.req_id; + if (!this.authed && !OPENING.has(msg.type)) { + refusedBeforeHandshake += 1; + return this.reply({ type: 'error', detail: 'Handshake required', req_id }, 1); + } + switch (msg.type) { + case 'handshake': + return this.reply({ type: 'handshake_challenge', v: '99.0', v_min: '0.0', + nonce: b64(new Uint8Array(32)), node_pk: 'pk', sig: 's', req_id }, rttMs); + case 'handshake_response': + setTimeout(() => { this.authed = true; }, rttMs); + return this.reply({ type: 'handshake_ack', proof: 'AA==', node_pk: 'pk', sig: 's', + nonce: 'n', ct: 'c', req_id }, rttMs); + case 'device_hello': + return this.reply({ type: 'device_hello_ack', req_id }, rttMs); + case 'file_req': + served += 1; + return this.reply({ type: 'file_chunk', file_id: msg.file_id, + chunk_index: msg.chunk_index, nonce: 1, ct: new Uint8Array([1]), req_id }, chunkMs); + default: + return undefined; + } + } +} + +class FakeChannel { + constructor() { this.readyState = 'connecting'; this.binaryType = 'arraybuffer'; } + send(frame) { + if (this.readyState !== 'open') throw new Error('InvalidStateError'); + this.session.receive(frame); + } + close() { + if (this.readyState === 'closed') return; + this.readyState = 'closed'; + if (this.onclose) setTimeout(() => this.onclose({}), lateCloseMs); + } +} + +class FakePC { + constructor() { + this.connectionState = 'new'; + this.iceConnectionState = 'new'; + this.iceGatheringState = 'complete'; + pcs.push(this); + } + createDataChannel() { this.channel = new FakeChannel(); return this.channel; } + async createOffer() { return { type: 'offer', sdp: SDP }; } + async setLocalDescription(d) { this.localDescription = d; } + async setRemoteDescription() { + setTimeout(() => { + if (this.connectionState === 'closed') return; + this.channel.session = new NodeSession(this.channel); + this.channel.readyState = 'open'; + this.connectionState = 'connected'; + if (this.channel.onopen) this.channel.onopen(); + }, openMs); + } + close() { + this.connectionState = 'closed'; + if (this.channel) this.channel.close(); + } +} +globalThis.RTCPeerConnection = FakePC; + +// ── The shipped transport and pipeline ─────────────────────────────────────── + +new Function(`${transportSrc}\nwindow.__codec = { msgpack_encode, msgpack_decode };`)(); +const Transport = window.MeshBayTransport; + +const constant = (name) => utils.match(new RegExp(`^const ${name} = [^;]+;`, 'm'))[0]; +const from = utils.indexOf('const CHUNK_RETRY_ATTEMPTS'); +const to = utils.indexOf('async function downloadEntry'); +const { pipelinedDownload } = new Function('window', 't', ` + ${constant('PIPELINE_WINDOW')} + ${constant('LANE_MIN_CHUNKS')} + ${utils.slice(from, utils.lastIndexOf('/**', to))} + return { pipelinedDownload };`)(window, (key) => key); + +const identity = { pkEdB64: 'ed', signAs: async () => 'sig', native: false }; +const main = new Transport('https://hub.invalid', 'token'); +let outcome = 'pending'; +let reconnected = 0; +main.addReconnectListener(() => { reconnected += 1; }); +const written = []; +const writable = { write: async (bytes) => { written.push(bytes[0]); } }; + +main.connect('node', 'token', 'group', new Uint8Array(32), identity, null, 'u', 'uid', + undefined, undefined, '', 'pk') + .then(() => pipelinedDownload(main, 'gek', 'f'.repeat(64), chunks, null, writable, null, '')) + .then(() => { outcome = 'done'; }, (err) => { outcome = `failed: ${err.message}`; }); + +await run(hangAt); +pcs[0].channel.session.hung = true; +await run(hangAt + failAfter); +pcs[0].connectionState = 'failed'; +pcs[0].onconnectionstatechange(); +await run(runUntil); + +process.stdout.write(JSON.stringify({ + outcome, reconnected, refusedBeforeHandshake, served, + written: written.length, + inOrder: written.every((b, i) => b === (i & 255)), + connections: pcs.length, +})); diff --git a/packages/meshbay-hub/tests/test_reconnect_race.py b/packages/meshbay-hub/tests/test_reconnect_race.py new file mode 100644 index 0000000..01c5182 --- /dev/null +++ b/packages/meshbay-hub/tests/test_reconnect_race.py @@ -0,0 +1,93 @@ +""" +A reconnect must not fail the download it is reconnecting for. + +Reported live on 0.20: a large download from a node under load stopped with +"Handshake required" in the transfers panel. `_sendAndWait` let every message +through while the reconnect's own connect() ran, because the handshake it sends +has to go through; so a chunk whose retry came due in that window went out on a +channel that had opened and not finished its handshake. The node answered +"Handshake required" (webrtc/dispatch.py), which no retry treats as a dropped +connection. Now only the opening messages pass, and anything else waits for the +session or fails as a dropped connection, which the download waits out. + +The replaced channel's close event is the second half: it reached the new +connection, marked it closed and failed what it had in flight. + +These run the shipped transport and pipeline against a node that refuses what +comes before the handshake: harness/reconnect_race_harness.mjs. +""" + +import json +import os +import re +import shutil +import subprocess +from pathlib import Path + +import pytest +from spa_source import classic_scripts, transport_source + +STATIC = Path(__file__).resolve().parents[1] / "src" / "meshbay_hub" / "static" +FILE_UTILS = STATIC / "file-utils.js" +HARNESS = Path(__file__).parent / "harness" / "reconnect_race_harness.mjs" +DISPATCH = (Path(__file__).resolve().parents[2] / "meshbay-node" / "src" / "meshbay_node" + / "transport" / "webrtc" / "dispatch.py") + +pytestmark = pytest.mark.skipif( + shutil.which("node") is None or not FILE_UTILS.exists(), + reason="node or the SPA sources are not available") + + +def _run(**cfg) -> dict: + transport = os.pathsep.join( + str(STATIC / name) for name in classic_scripts() if name.startswith("transport")) + proc = subprocess.run( + ["node", str(HARNESS), str(FILE_UTILS), transport, json.dumps(cfg)], + capture_output=True, text=True, encoding="utf-8", timeout=120) + assert proc.returncode == 0, proc.stderr + return json.loads(proc.stdout) + + +def test_retries_that_come_due_during_the_handshake_wait_for_it(): + # The chunks in flight time out just before the connection is declared + # failed, and their retries land while the new channel is handshaking. + out = _run() + assert out["reconnected"] == 1 and out["connections"] == 2 + assert out["refusedBeforeHandshake"] == 0, "a chunk was asked for before the handshake" + assert out["outcome"] == "done" + assert out["written"] == 400 and out["inOrder"] + + +@pytest.mark.parametrize("cfg", [ + {"failAfter": 20000}, # failure declared before anything times out + {"failAfter": 31000, "rttMs": 800}, # a slow handshake + {"openMs": 2500}, # a slow channel +]) +def test_the_download_survives_other_timings(cfg): + out = _run(**cfg) + assert out["refusedBeforeHandshake"] == 0 + assert out["outcome"] == "done" and out["written"] == 400 and out["inOrder"] + + +@pytest.mark.parametrize("late", [800, 1500, 3000]) +def test_the_replaced_channels_close_event_is_ignored(late): + out = _run(lateCloseMs=late) + assert out["outcome"] == "done" and out["written"] == 400 + + +def _node_opening_types() -> set[str]: + """What the node answers before the handshake, read from its dispatch.""" + from meshbay_common.protocol import MNP + src = DISPATCH.read_text(encoding="utf-8") + before = src[src.index("def _dispatch_message"):src.index("elif self._user_id is None")] + names = set(re.findall(r"MNP\.([A-Z_]+)", before)) + return {getattr(MNP, n) for n in names} + + +def test_the_opening_messages_are_the_ones_the_node_takes_before_the_handshake(): + m = re.search(r"const SESSION_OPENING = new Set\(\[(.*?)\]\);", transport_source(), re.S) + client = set(re.findall(r"'([a-z_]+)'", m.group(1))) + # device_hello is the one connect() sends after the proof and before it + # returns: the node takes it then, and the session is not up until it has. + assert client - _node_opening_types() == {"device_hello"} + assert _node_opening_types() <= client |