diff options
Diffstat (limited to 'packages/meshbay-hub')
3 files changed, 45 insertions, 8 deletions
diff --git a/packages/meshbay-hub/src/meshbay_hub/static/transport-media.js b/packages/meshbay-hub/src/meshbay_hub/static/transport-media.js index ca015b9b..91af0338 100644 --- a/packages/meshbay-hub/src/meshbay_hub/static/transport-media.js +++ b/packages/meshbay-hub/src/meshbay_hub/static/transport-media.js @@ -254,7 +254,13 @@ extendTransport(class { /** Room for `n` more segments. */ grantStreamCredit(n = 1) { - if (!this._connected) return; + // Only on a session that is up: the player goes on granting as it plays + // from its buffer through a reconnect, and a grant that reached the new + // channel before its handshake was answered "Handshake required", which + // the player showed in place of the film. The stream went with the lost + // session; the reconnect asks for it again (video-player.js + // `reconnectPlan`). The same goes for the diagnostic and the stop below. + if (!this._sessionUp) return; console.log('[stream] grant credit:', n); this._send({ type: 'stream_more', v: '0.1', n }); } @@ -267,7 +273,7 @@ extendTransport(class { * the two halves in one file. The node only logs it. */ sendStreamDiag(diag) { - if (!this._connected) return; + if (!this._sessionUp) return; try { this._send({ type: 'client_diag', v: '0.1', ...diag }); } catch { /* gone */ } } @@ -279,7 +285,7 @@ extendTransport(class { * the next video answered "server busy". */ stopStream() { - if (!this._connected) return; + if (!this._sessionUp) return; try { this._send({ type: 'stream_stop', v: '0.1' }); } catch { /* gone */ } } }); diff --git a/packages/meshbay-hub/tests/harness/reconnect_race_harness.mjs b/packages/meshbay-hub/tests/harness/reconnect_race_harness.mjs index 0f1cd979..606598da 100644 --- a/packages/meshbay-hub/tests/harness/reconnect_race_harness.mjs +++ b/packages/meshbay-hub/tests/harness/reconnect_race_harness.mjs @@ -30,6 +30,7 @@ const { rttMs = 300, // one request → its reply, during the handshake lateCloseMs = 0, // >0: the replaced channel's close event fires this late runUntil = 300000, + stream = false, // a film playing instead of a download (see below) } = JSON.parse(process.argv[4] || '{}'); for (const level of ['log', 'warn', 'error', 'info', 'debug']) { @@ -106,6 +107,7 @@ globalThis.fetch = async (url) => ({ const OPENING = new Set(['handshake', 'handshake_response', 'keypair_bundle_fetch', 'gek_bundle_fetch', 'join_request', 'device_add_request', 'device_hello']); let refusedBeforeHandshake = 0; +const refusedTypes = new Set(); let served = 0; const pcs = []; @@ -127,6 +129,7 @@ class NodeSession { const req_id = msg.req_id; if (!this.authed && !OPENING.has(msg.type)) { refusedBeforeHandshake += 1; + refusedTypes.add(msg.type); return this.reply({ type: 'error', detail: 'Handshake required', req_id }, 1); } switch (msg.type) { @@ -210,10 +213,26 @@ 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}`; }); +// A film playing (`stream`): the player goes on granting the node room for +// segments as it plays from its buffer, and on telling it what it sees, through +// the connection loss and the reconnect, as video-player.js does. A refusal of +// either is a node error the player shows in place of the film. +const opened = main.connect('node', 'token', 'group', new Uint8Array(32), identity, null, + 'u', 'uid', undefined, undefined, '', 'pk'); +if (stream) { + opened.then(() => { + main.requestStream('f'.repeat(64)); + setInterval(() => { + main.grantStreamCredit(1); + main.sendStreamDiag({ buffered: 1 }); + }, 50); + outcome = 'done'; + }, (err) => { outcome = `failed: ${err.message}`; }); +} else { + opened + .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; @@ -223,7 +242,7 @@ pcs[0].onconnectionstatechange(); await run(runUntil); process.stdout.write(JSON.stringify({ - outcome, reconnected, refusedBeforeHandshake, served, + outcome, reconnected, refusedBeforeHandshake, refusedTypes: [...refusedTypes], 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 index 01c51824..539ae268 100644 --- a/packages/meshbay-hub/tests/test_reconnect_race.py +++ b/packages/meshbay-hub/tests/test_reconnect_race.py @@ -75,6 +75,18 @@ def test_the_replaced_channels_close_event_is_ignored(late): assert out["outcome"] == "done" and out["written"] == 400 +@pytest.mark.parametrize("cfg", [{}, {"rttMs": 800}, {"openMs": 2500}]) +def test_a_film_playing_says_nothing_before_the_handshake(cfg): + # Reported live on a phone moving from mobile data to Wi-Fi: the player + # went on granting stream credit through the reconnect, one grant reached + # the new channel before its handshake, and the node's "Handshake required" + # replaced the film. A grant or a diagnostic for a stream the lost session + # took with it has nowhere to go; the reconnect asks for the stream again. + out = _run(stream=True, runUntil=60000, **cfg) + assert out["outcome"] == "done" and out["reconnected"] == 1 + assert out["refusedTypes"] == [], out["refusedTypes"] + + def _node_opening_types() -> set[str]: """What the node answers before the handshake, read from its dispatch.""" from meshbay_common.protocol import MNP |