/** * Transfers that outlive the page that started them. * * Downloads and uploads used to be state inside GroupPage, which meant leaving * a group killed them — the component unmounted, its effect closed the * DataChannel, and a half-written file was all you had. They live here instead: * a module-level store that nothing unmounts, with the group page as one of * several possible views onto it. * * Two consequences worth stating, because they are the reason this exists: * * - The transport cannot be closed just because a page went away. A group * page hands its transport over with `releaseWhenIdle()`, and the last * transfer using it closes it. * - Signing out is different from navigating. It cancels everything and * closes what it was using, because the tokens those transfers are running * on are about to stop being ours. * * No browser globals: exercised under Node by * packages/meshbay-hub/tests/test_transfers.py. */ const SPEED_WINDOW_MS = 5000; function _abortError() { const err = new Error('Cancelled'); err.name = 'AbortError'; return err; } let _nextId = 1; export class TransferStore { constructor(now = () => Date.now()) { this._now = now; this._items = []; this._subs = new Set(); this._releasing = new Set(); } subscribe(fn) { this._subs.add(fn); return () => this._subs.delete(fn); } _emit() { for (const fn of this._subs) fn(this.list()); } /** * What a view needs to render, as plain data — never the internals, so a * render cannot accidentally hold a transport alive. */ list() { return this._items.map(it => ({ id: it.id, kind: it.kind, name: it.name, total: it.total, done: it.done, status: it.status, // How many are in front of this one, and whose limit is holding it up: // "your own two slots are busy" and "the node is full" are different // situations and the person can act on only one of them. ahead: it.ahead || 0, queuedByOwnLimit: Boolean( it.lease && it.lease.cap && it.lease.used >= it.lease.cap), error: it.error || '', speed: this._speed(it), // The ETA is drawn only once the window holds a few seconds of real // measurement -- see etaSeconds. settled: it.samples.length > 2 && (it.samples[it.samples.length - 1].t - it.samples[0].t) >= 3000, percent: it.total ? Math.min(100, Math.round(it.done / it.total * 100)) : 0, // Only for a file written into a folder the browser granted us: that is // the one case where the page can read its own download back. canOpen: it.status === 'done' && typeof it.open === 'function', })); } get active() { return this._items.filter(it => it.status === 'running').length; } /** Running or waiting for a slot — what the nav badge counts. */ get pending() { return this._items.filter( it => it.status === 'running' || it.status === 'queued').length; } _speed(it) { // Over a window rather than since the start: a transfer that stalls should // read as slow immediately, not as its own historical average. const s = it.samples; if (s.length < 2) return 0; const dt = (s[s.length - 1].t - s[0].t) / 1000; if (dt <= 0) return 0; return (s[s.length - 1].done - s[0].done) / dt; } /** * Start a transfer. * * `run` receives `{ signal, onProgress }`. It must poll `signal.aborted` — a * cancel that only sets a flag nobody reads is a button that lies. */ start({ kind, name, total = 0, transport = null, run, open = null, lease = null }) { const item = { id: _nextId++, kind, name, total, transport, open, lease, done: 0, // A transfer that has to wait for a slot starts as 'queued', not // 'running'. Two different things are true of it — nothing is moving, and // nothing is wrong — and a status that conflates them is what makes a // queue look like a hang. status: lease && lease.state !== 'granted' ? 'queued' : 'running', ahead: (lease && lease.ahead) || 0, error: '', samples: [{ t: this._now(), done: 0 }], signal: { aborted: false }, }; this._items.push(item); this._emit(); const onProgress = (done, total) => { item.done = done; if (total) item.total = total; const t = this._now(); item.samples.push({ t, done }); while (item.samples.length > 2 && t - item.samples[0].t > SPEED_WINDOW_MS) { item.samples.shift(); } this._emit(); }; const finish = (status, error = '') => { item.status = status; item.error = error; this._emit(); this._maybeRelease(item.transport); }; // The slot is given back in a `finally` around everything, so it survives // a throw, a cancel and a return alike. A slot not returned is a member who // cannot start another transfer until the node times it out. const finished = () => { if (item.lease) item.lease.release( item.signal.aborted ? 'cancelled' : 'done'); }; // Installed here and not inside the promise chain below. A state push that // arrived before the first microtask ran was simply dropped, so a transfer // could sit at the position it was given when it was created and never // appear to move — the widget showing "3 ahead" for ever while the node // quietly worked through the queue. Nothing about that looks wrong from // either side, which is why it needs a test rather than a reading. if (item.lease) { item.lease._onState = (lease) => { if (item.status !== 'queued' && item.status !== 'running') return; item.ahead = lease.ahead; item.status = lease.state === 'granted' ? 'running' : 'queued'; this._emit(); }; } const promise = Promise.resolve() .then(async () => { if (item.lease) { await item.lease.acquire(); if (item.signal.aborted) throw _abortError(); item.status = 'running'; this._emit(); } return run({ signal: item.signal, onProgress, lease: item.lease }); }) .then(() => { if (item.signal.aborted) finish('cancelled'); else { if (item.total) item.done = item.total; finish('done'); } }) .catch(err => { if (item.signal.aborted || err.name === 'AbortError') finish('cancelled'); else finish('failed', err.message || String(err)); }) .finally(finished); item.promise = promise; return item.id; } /** * Hand a finished download to the browser to display. * * As close to "open it" as a web page gets: the bytes go to a new tab and the * browser decides what to do with them. A page cannot start a desktop * application, and cannot show a file manager — there is no API for either, * in any browser, by design. */ open(id) { const item = this._items.find(it => it.id === id); if (item && typeof item.open === 'function') return item.open(); } cancel(id) { const item = this._items.find(it => it.id === id); // 'queued' too: a transfer waiting for a slot is exactly the one somebody // is most likely to give up on, and its queue entry has to go with it or // the node grants a slot to a transfer that will never use it. if (!item || (item.status !== 'running' && item.status !== 'queued')) return; item.signal.aborted = true; if (item.lease) item.lease.release('cancelled'); // Marked at once. The work stops when it next looks, but a cancelled // transfer should not keep reporting progress in the meantime. item.status = 'cancelled'; this._emit(); this._maybeRelease(item.transport); } cancelAll() { for (const it of this._items) { if (it.status === 'running' || it.status === 'queued') this.cancel(it.id); } } /** Drop everything finished, keeping what is still running or waiting. */ clearFinished() { this._items = this._items.filter( it => it.status === 'running' || it.status === 'queued'); this._emit(); } _busy(transport) { // Queued counts as busy: a transport closed while a transfer waits for a // slot can never be granted one, and the transfer would sit at "waiting" // for ever with nothing left to answer it. return this._items.some( it => it.transport === transport && (it.status === 'running' || it.status === 'queued')); } /** * The group page is going away. Close its transport once nothing is using it, * which may be now or may be in twenty minutes. */ releaseWhenIdle(transport) { if (!transport) return; this._releasing.add(transport); this._maybeRelease(transport); } _maybeRelease(transport) { if (!transport || !this._releasing.has(transport)) return; if (this._busy(transport)) return; this._releasing.delete(transport); try { transport.onIndexSync = null; transport.close(); } catch { /* already gone */ } } /** Signing out: stop everything and let go of every transport. */ reset() { this.cancelAll(); for (const transport of [...this._releasing]) { this._releasing.delete(transport); try { transport.close(); } catch { /* already gone */ } } for (const it of this._items) { if (it.transport) { try { it.transport.close(); } catch { /* already gone */ } } } this._items = []; this._emit(); } } export const transfers = new TransferStore(); /** * Seconds left, or null when saying nothing is the honest answer. * * Withheld until the speed window has real samples in it: a figure computed * from the first two chunks of a transfer swings between "4 seconds" and "an * hour" and back, and a number that behaves like that is worse than a blank — * people read the first one they see and plan around it. */ export function etaSeconds(item) { if (item.status !== 'running' || !item.total || !item.speed) return null; const left = item.total - item.done; if (left <= 0) return null; const secs = left / item.speed; return Number.isFinite(secs) ? secs : null; } /** Human-readable rate, for a widget that updates several times a second. */ export function formatSpeed(bytesPerSecond) { if (!bytesPerSecond || bytesPerSecond < 1) return ''; if (bytesPerSecond < 1024 * 1024) return `${Math.round(bytesPerSecond / 1024)} KB/s`; return `${(bytesPerSecond / (1024 * 1024)).toFixed(1)} MB/s`; }