/** * 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; /** Not finished: still preparing, waiting for a slot, or transferring. One * definition, because six places ask and they were drifting apart. */ function _live(status) { return status === 'preparing' || status === 'queued' || status === 'running' || status === 'paused'; } /** Raised by `run` when it stopped because the transfer was paused. */ function _pausedError() { const err = new Error('Paused'); err.name = 'PausedError'; return err; } 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', // Whether the target can be stopped and continued. False is the honest // answer for a service-worker stream, and the button is not drawn. pausable: Boolean(it.pausable), })); } 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 => _live(it.status)).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 a transfer. * * `run` receives `{ signal, onProgress, lease }`. It must poll * `signal.aborted` — a cancel that only sets a flag nobody reads is a button * that lies. * * `prepare` is optional and runs before anything else, with the row already * on screen. It is where a download opens its target, which can take tens of * seconds — the streamed path waits for the worker, twice, and a Save As * dialog waits for a person. Doing that *before* creating the row meant three * clicks produced no panel at all, not even the icon, and then several rows * at once. Returning `false` drops the row again, which is what a dismissed * dialog should look like: nothing, rather than a cancelled transfer nobody * started. * * `makeLease` is called after `prepare` succeeds, never before. A granted * slot must be taken up within the node's deadline, so it is asked for once * there is somewhere to write — see file-utils.js's downloadEntry. */ start({ kind, name, total = 0, transport = null, run, open = null, lease = null, prepare = null, makeLease = null, pausable = false }) { 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: prepare ? 'preparing' : (lease && lease.state !== 'granted' ? 'queued' : 'running'), ahead: (lease && lease.ahead) || 0, error: '', samples: [{ t: this._now(), done: 0 }], signal: { aborted: false, paused: false }, // Whether this transfer can be stopped and continued. A download learns // it from `prepare`, because only its target knows; an upload says so // outright, because a `File` is always seekable and the node keeps the // position (see uploads.py). pausable: Boolean(pausable), // Where a resumed run picks up, in chunks. Zero until something pauses. resumeFrom: 0, // Resolved by resume(); awaited by the run loop while paused. resumed: null, _wake: null, // Pausing gives the slot back, so resuming has to be able to ask for // another one. A transfer handed a lease directly cannot, and must not // be offered a button that would drop its slot for good. _canRelease: Boolean(makeLease), }; 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) this._watchLease(item); const promise = Promise.resolve() .then(async () => { if (prepare) { const ready = await prepare(); if (item.signal.aborted) throw _abortError(); if (ready === false) { // Dismissed. Not a failure and not a cancellation: nothing was ever // started, so nothing should be left on screen to explain. this._drop(item.id); return undefined; } if (ready && ready.name) item.name = ready.name; // Only the target knows. A service-worker stream is already an HTTP // response the browser is writing to its own download folder: not // writing to it stalls that download outside our control, and an idle // worker is terminated within seconds, taking the stream with it. So // the button is offered where it works and nowhere else — a pause // that silently restarts from zero is worse than no pause. if (ready && ready.pausable) item.pausable = true; item.status = 'running'; this._emit(); } // Run, and be prepared to be stopped and started again. // // A paused transfer holds **nothing**: its slot goes back to the node // and resuming rejoins the queue at the tail. Anything else lets one // member close a node by pausing four downloads and going to lunch. // So the lease is taken inside this loop, not before it. for (;;) { if (makeLease && !item.lease) { item.lease = makeLease(); this._watchLease(item); if (item.lease.state !== 'granted') { item.status = 'queued'; item.ahead = item.lease.ahead || 0; this._emit(); } } if (item.lease) { await item.lease.acquire(); if (item.signal.aborted) throw _abortError(); item.status = 'running'; this._emit(); } try { return await run({ signal: item.signal, onProgress, lease: item.lease, from: item.resumeFrom || 0 }); } catch (err) { if (err.name !== 'PausedError') throw err; } // Where to pick up. `run` records it on the signal rather than // returning it, because it has to survive being thrown past. item.resumeFrom = item.signal.resumeFrom || 0; if (item.lease) { item.lease.release('paused'); item.lease = null; } item.status = 'paused'; item.ahead = 0; this._emit(); this._maybeRelease(item.transport); await item.resumed; if (item.signal.aborted) throw _abortError(); } }) .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; } /** Resolves with a transfer's final status once it has ended, whatever the ending. */ settled(id) { const item = this._items.find((it) => it.id === id); return item ? item.promise.then(() => item.status) : Promise.resolve('gone'); } _watchLease(item) { 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(); }; } /** Remove a row entirely. Only for a transfer that never started. */ _drop(id) { this._items = this._items.filter(it => it.id !== id); this._emit(); } /** * 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(); } /** * Stop a running transfer, keeping what it has already written. * * Only while running: a queued transfer is already stopped and holds no slot, * and pausing it would only cost it its place. Only where the target can do * it — see the note in `start`. * * The slot goes back to the node at once (§6.2 of the plan): a paused * transfer holds nothing, and resuming rejoins the queue at the tail. */ pause(id) { const item = this._items.find(it => it.id === id); if (!item || !item.pausable || !item._canRelease || item.status !== 'running') return; item.signal.paused = true; // Created here rather than in resume(): the run loop awaits it the moment // `run` throws, which can be sooner than the next call into this store. item.resumed = new Promise((resolve) => { item._wake = resolve; }); this._emit(); } /** Start it again, from where it stopped, behind whatever is waiting now. */ resume(id) { const item = this._items.find(it => it.id === id); if (!item || item.status !== 'paused') return; item.signal.paused = false; item.status = 'queued'; this._emit(); if (item._wake) { item._wake(); item._wake = null; } } 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 || !_live(item.status)) return; item.signal.aborted = true; if (item.lease) item.lease.release('cancelled'); // A paused run is parked on `item.resumed`. Without this it stays parked // for the life of the page, holding its target open, and the row says // "cancelled" over a download that never stopped. if (item._wake) { item._wake(); item._wake = null; } // 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 (_live(it.status)) this.cancel(it.id); } } /** Drop everything finished, keeping what is still running or waiting. */ clearFinished() { this._items = this._items.filter(it => _live(it.status)); 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 && _live(it.status)); } /** * 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`; }