diff options
Diffstat (limited to 'packages/meshbay-hub/src/meshbay_hub/static/transport.js')
| -rw-r--r-- | packages/meshbay-hub/src/meshbay_hub/static/transport.js | 213 |
1 files changed, 211 insertions, 2 deletions
diff --git a/packages/meshbay-hub/src/meshbay_hub/static/transport.js b/packages/meshbay-hub/src/meshbay_hub/static/transport.js index c4a24c5..54a1302 100644 --- a/packages/meshbay-hub/src/meshbay_hub/static/transport.js +++ b/packages/meshbay-hub/src/meshbay_hub/static/transport.js @@ -283,6 +283,129 @@ const JOIN_REFUSALS = { group_mismatch: 'The node refused a request naming a different group.', }; +/** + * One transfer's slot on the node, from this side. + * + * The contract the transfer store depends on: `acquire()` resolves when the + * node has granted the slot (immediately on a node that hands out none), and + * `release(reason)` gives it back exactly once. Nothing else in the client + * speaks to the node about slots. + * + * Two things here exist only because a queue can lie, and both are the + * difference between "waiting" and "waiting for ever": + * + * - **the watchdog.** A grant is pushed, not polled, so a lost push leaves + * this side waiting on a node that believes it has started. Re-asking is + * free — the node is idempotent on `tr` — and it is the only thing that + * recovers a message that did not arrive. + * - **release is idempotent and unconditional.** A slot given back twice + * costs nothing; one never given back is a member who cannot transfer + * again until a timeout the node runs on its own. + */ +const LEASE_WATCHDOG_MS = 60000; + +class Lease { + constructor(transport, tr, kind, bytes, chunks, onState) { + this.transport = transport; + this.tr = tr; + this.kind = kind; + this.bytes = bytes; + this.chunks = chunks; + this.state = 'opening'; + this.ahead = 0; + this.closed = false; + this._onState = onState; + this._granted = null; + this._watchdog = 0; + this._wait = new Promise((resolve) => { this._granted = resolve; }); + } + + /** No slots on this node: behave as though one was granted at once. */ + _skip() { + this.state = 'granted'; + this._granted(); + } + + _request() { + // A closed channel is not a failure here, and must not throw: the transport + // reconnects on its own, `_reopenTransfers` re-asks for every live lease + // when it does, and the watchdog below asks again meanwhile. + // + // This is the same tolerance `_fetchChunkResilient` already gives a chunk + // request — and before leases existed, a chunk request was the first thing + // to touch the channel, so a download started on a briefly dead connection + // simply retried. Asking for a slot first made `_send` the first contact + // and threw "DataChannel not open (state: closed)" out of `downloadEntry`, + // where nothing catches it: a download that used to recover became an + // error with no row in the widget to show it. Found by downloading a file + // right after a connection dropped. + try { + this.transport._send({ + type: 'transfer_open', v: '0.1', tr: this.tr, kind: this.kind, + bytes: this.bytes, chunks: this.chunks, + }); + } catch (err) { + console.warn('[MeshBay] could not ask for a slot yet:', err.message); + } + this._arm(); + } + + _arm() { + clearTimeout(this._watchdog); + if (this.closed || this.state === 'granted') return; + this._watchdog = setTimeout(() => { + if (this.closed || this.state === 'granted') return; + console.warn('[MeshBay] no answer for transfer', this.tr.slice(0, 8), + '- asking again'); + this._request(); + }, LEASE_WATCHDOG_MS); + } + + _apply(msg) { + if (this.closed) return; + this.state = msg.state; + this.ahead = msg.ahead || 0; + this.used = msg.used; + this.cap = msg.cap; + if (msg.state === 'granted') { + clearTimeout(this._watchdog); + this._granted(); + } else if (msg.state === 'closed') { + // The node ended it: reclaimed as idle, or revoked. Not an error here — + // whoever is running the transfer finds out through its own failure — but + // the slot is gone and asking again is the only way back. + clearTimeout(this._watchdog); + } else { + this._arm(); + } + if (this._onState) { + try { this._onState(this); } catch (e) { + console.error('[MeshBay] lease state handler threw:', e); + } + } + } + + /** Resolves once the node has granted the slot. */ + acquire() { return this._wait; } + + /** + * Give the slot back. Safe to call twice, and safe on a dead transport: a + * lease that is not released is a member who cannot start another transfer + * until the node times it out, so this must never be conditional on anything. + */ + release(reason = 'done') { + if (this.closed) return; + this.closed = true; + clearTimeout(this._watchdog); + this.transport._leases.delete(this.tr); + if (!this.transport.supportsTransferSlots) return; + try { + this.transport._send({ type: 'transfer_close', v: '0.1', tr: this.tr, + reason }); + } catch { /* the connection is gone, and so is the lease with it */ } + } +} + class MeshBayTransport { constructor(hubUrl, accessToken) { this._hubUrl = hubUrl; @@ -315,6 +438,13 @@ class MeshBayTransport { // Names, not ids: the "already being uploaded" guard is about the file the // caller passed, and two `uploadFile` calls for one file draw two ids. this._inFlightUploads = new Set(); + // tr → Lease. A transfer's slot on the node, from the client's side. + this._leases = new Map(); + // Set from the handshake ack: a node that answers with `transfer_limits` + // speaks transfer slots. Used instead of a timeout, because "no answer + // yet" and "this node will never answer" are indistinguishable in time and + // guessing wrong either stalls every download or defeats the cap. + this._transferLimits = null; // 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 @@ -377,6 +507,20 @@ class MeshBayTransport { get nodeVersion() { return this._nodeVersion || ''; } /** + * Whether this node hands out transfer slots. + * + * Read from the handshake ack rather than from the MNP version: the caps + * shipped before the version bump that will make leases compulsory, so for + * now a node either answers with `transfer_limits` or it predates all of + * this. A node that does not is asked for nothing and enforces nothing — + * every download behaves exactly as it did. + */ + get supportsTransferSlots() { return this._transferLimits !== null; } + + /** This member's own caps in this group, or null when the node said nothing. */ + get transferLimits() { return this._transferLimits; } + + /** * Whether the node speaks the per-root and per-app operations MNP 1.1 added: * `root_update`/`root_eject`/`root_plug`, `app_directories`, * `chat_directory`, `chat_link_preview`. @@ -880,6 +1024,7 @@ class MeshBayTransport { delete ack.nonce; delete ack.ct; Object.assign(ack, config); + this._transferLimits = ack.transfer_limits || null; // Tell the node which of this account's devices is on this connection. // Deliberately after the ack, and gated on the node's own version rather @@ -1016,6 +1161,10 @@ class MeshBayTransport { } trace('reconnect_ok', { attempt: this._reconnectAttempts }); console.log('[MeshBay] Reconnected after', this._reconnectAttempts, 'attempt(s)'); + // Before the caller's own hook: a transfer that resumes mid-chunk must + // have asked for its slot back first, or its next `file_req` carries a + // `tr` the node has never heard of. + this._reopenTransfers(); if (this._onReconnected) { try { this._onReconnected(); } catch (e) { console.error('[MeshBay] onReconnected handler threw:', e); @@ -1100,12 +1249,52 @@ class MeshBayTransport { return msg; } - async fetchChunk(fileId, chunkIndex) { + // ── Transfer slots ───────────────────────────────────────────────────────── + + /** + * Ask the node for a slot, and wait until it says yes. + * + * `tr` is drawn here, not by the node — 16 random bytes, exactly like + * `upload_id` — which is what makes re-opening after a reconnect idempotent + * rather than a second charge against the member's cap. + * + * On a node that predates transfer slots this resolves at once and costs + * nothing: there is no cap to respect and no message that would be + * understood. + */ + openTransfer({ kind = 'download', bytes = 0, chunks = 0, onState = null } = {}) { + const tr = _hex(crypto.getRandomValues(new Uint8Array(16))); + const lease = new Lease(this, tr, kind, bytes, chunks, onState); + if (!this.supportsTransferSlots) { + lease._skip(); + return lease; + } + this._leases.set(tr, lease); + lease._request(); + return lease; + } + + /** Re-ask for every live lease. Called after a reconnect. */ + _reopenTransfers() { + if (!this.supportsTransferSlots) return; + for (const lease of this._leases.values()) { + // The node lost the lease with the session, so this is a fresh request + // for the same `tr` — which the node treats as the same transfer rather + // than a second one. + if (!lease.closed) lease._request(); + } + } + + async fetchChunk(fileId, chunkIndex, tr = '') { const msg = await this._sendAndWait({ type: 'file_req', v: '0.1', file_id: fileId, chunk_index: chunkIndex, + // Present only when this download holds a slot. The node does not require + // it yet; carrying it is what lets the node see the transfer is alive and + // not reclaim its slot as idle. + ...(tr ? { tr } : {}), }); if (msg.type === 'error') throw new Error(msg.detail); return msg; @@ -2189,7 +2378,8 @@ class MeshBayTransport { * folder on screen to name. Omitting both leaves the node to pick, which it * only does for a client old enough to have had one destination. */ - async uploadFile(file, { chunkSize, onProgress, signal, root, dir } = {}) { + async uploadFile(file, { chunkSize, onProgress, signal, root, dir, + tr = '' } = {}) { // The same file twice at once would confuse the node, which keys its own // upload state by name — and would race for the same destination. The guard // is by name for that reason, even though the map below is keyed by id. @@ -2273,6 +2463,7 @@ class MeshBayTransport { upload_id: uploadId, chunk_index: i, total_chunks: total, + ...(tr ? { tr } : {}), ...sealed, }); } @@ -2992,6 +3183,24 @@ class MeshBayTransport { // "arrived" — and every message after that is one slot off too. Found // live: a group mid-scan corrupted its own handshake and chat history // this way, arriving roughly every 2s for as long as scanning ran. + // Routed by `tr`, and only by `tr`. A grant arrives unsolicited, minutes + // after the request that produced it, so falling through to "the oldest + // pending request" would hand a chat send or a handshake somebody else's + // slot — the class of defect `req_id` was introduced for. + if (msg.type === 'transfer_state') { + const lease = this._leases.get(msg.tr); + if (lease) lease._apply(msg); + else if (msg.state === 'granted') { + // A grant for a transfer this page has forgotten (a reload, a cancel + // that raced the grant). Handing it back at once matters: otherwise the + // node holds it until the 30 s acceptance deadline, and everyone behind + // it waits for nothing. + this._send({ type: 'transfer_close', v: '0.1', tr: msg.tr, + reason: 'cancelled' }); + } + return; + } + if (msg.type === 'index_progress') { if (this._onIndexProgress) { this._onIndexProgress({ |