diff --git a/src/puter-js/src/modules/networking/PSocket.js b/src/puter-js/src/modules/networking/PSocket.js index 8ac2edb5e..6d10bd887 100644 --- a/src/puter-js/src/modules/networking/PSocket.js +++ b/src/puter-js/src/modules/networking/PSocket.js @@ -1,5 +1,5 @@ import EventListener from '../../lib/EventListener.js'; -import { clearEpoxyClientCache, getEpoxyClient } from './index.js'; +import { discardOnRelayFailure, getEpoxyClient } from './index.js'; const textEncoder = new TextEncoder(); @@ -58,6 +58,7 @@ export class PSocket extends EventListener { #port; #useTls; + #client; #reader; #writer; @@ -65,6 +66,8 @@ export class PSocket extends EventListener { #closing = false; #closed = false; #pendingWrites = []; + #flushing = false; + #errored = false; /** * @param {string} host hostname or IP address of the server to connect to @@ -130,7 +133,8 @@ export class PSocket extends EventListener { throw new Error('Socket is already closed.'); } - if ( ! this.#writer ) { + // Writes made while the queue is still flushing go behind it. + if ( !this.#writer || this.#flushing ) { this.#pendingWrites.push({ payload, callback }); return; } @@ -154,24 +158,31 @@ export class PSocket extends EventListener { async #connect () { try { - await this.#connectWithClient(false); - } catch { + await this.#connectWithClient(); + } catch ( error ) { + // Retry only when the client could not be built or its relay is + // dead; a refused port or failed TLS handshake would fail again. + if ( this.#client && !discardOnRelayFailure(this.#client, error) ) { + this.#emitErrorAndClose(error); + return; + } try { - await this.#connectWithClient(true); + await this.#connectWithClient(); } catch ( retryError ) { - clearEpoxyClientCache(); + discardOnRelayFailure(this.#client, retryError); this.#emitErrorAndClose(retryError); } } } - async #connectWithClient (refresh) { + async #connectWithClient () { if ( this.#closing || this.#closed ) { return; } - const client = await getEpoxyClient({ refresh }); - const stream = await this.#openStream(client); + this.#client = undefined; + this.#client = await getEpoxyClient(); + const stream = await this.#openStream(this.#client); if ( this.#closing || this.#closed ) { try { @@ -191,8 +202,10 @@ export class PSocket extends EventListener { this.#writer = stream.write.getWriter(); this.#open = true; - this.emit(this.#eventName('open')); await this.#flushPendingWrites(); + if (this.#closed || this.#closing) return; + // A throwing handler must not read as a failed connect and retry. + this.#emitFromListener(this.#eventName('open')); void this.#readLoop(); } @@ -205,9 +218,16 @@ export class PSocket extends EventListener { } async #flushPendingWrites () { - while ( this.#pendingWrites.length && !this.#closed && !this.#closing ) { - const { payload, callback } = this.#pendingWrites.shift(); - await this.#writePayload(payload, callback); + this.#flushing = true; + try { + while ( this.#pendingWrites.length && !this.#closed && !this.#closing ) { + const { payload, callback } = this.#pendingWrites.shift(); + await this.#writePayload(payload, callback); + } + } finally { + // Cleared in the same tick as the last queue check; a write + // landing before the caller resumes would otherwise stay queued. + this.#flushing = false; } } @@ -226,7 +246,7 @@ export class PSocket extends EventListener { } } } catch ( error ) { - clearEpoxyClientCache(); + discardOnRelayFailure(this.#client, error); this.#emitErrorAndClose(error); } } @@ -261,7 +281,7 @@ export class PSocket extends EventListener { return; } - clearEpoxyClientCache(); + discardOnRelayFailure(this.#client, error); this.#emitErrorAndClose(error); } finally { try { @@ -309,15 +329,27 @@ export class PSocket extends EventListener { } #emitErrorAndClose (reason) { - if ( this.#closed ) { + if ( this.#closed || this.#errored ) { return; } - this.emit('error', normalizeError(reason)); + this.#errored = true; this.#closing = true; + this.#emitFromListener('error', normalizeError(reason)); void this.#closeStreams(true); } + // Emits on an internal path, surfacing a listener's throw on its own tick. + #emitFromListener (event, ...args) { + try { + this.emit(event, ...args); + } catch ( listenerError ) { + setTimeout(() => { + throw listenerError; + }, 0); + } + } + #emitClose (hadError) { if ( this.#closed ) { return; diff --git a/src/puter-js/src/modules/networking/PSocket.test.js b/src/puter-js/src/modules/networking/PSocket.test.js index cf4bc86cb..4a638c046 100644 --- a/src/puter-js/src/modules/networking/PSocket.test.js +++ b/src/puter-js/src/modules/networking/PSocket.test.js @@ -4,10 +4,10 @@ import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; // no wasm runtime, no relay, no websocket. Stubbing it also breaks the // PSocket <-> index.js import cycle for these tests. const mockGetEpoxyClient = vi.fn(); -const mockClearEpoxyClientCache = vi.fn(); +const mockDiscardOnRelayFailure = vi.fn(); vi.mock('./index.js', () => ({ getEpoxyClient: (...args) => mockGetEpoxyClient(...args), - clearEpoxyClientCache: (...args) => mockClearEpoxyClientCache(...args), + discardOnRelayFailure: (...args) => mockDiscardOnRelayFailure(...args), })); const { PSocket, PTLSSocket } = await import('./PSocket.js'); @@ -132,7 +132,7 @@ async function connected (stream = makeStream()) { beforeEach(() => { mockGetEpoxyClient.mockReset(); - mockClearEpoxyClientCache.mockReset(); + mockDiscardOnRelayFailure.mockReset().mockReturnValue(false); }); afterEach(() => { @@ -145,7 +145,7 @@ describe('PSocket connect', () => { expect(client.connect).toHaveBeenCalledWith('example.com', 80); expect(client.connectTls).not.toHaveBeenCalled(); - expect(mockGetEpoxyClient).toHaveBeenCalledWith({ refresh: false }); + expect(mockGetEpoxyClient).toHaveBeenCalledTimes(1); expect(events.error).not.toHaveBeenCalled(); }); @@ -160,21 +160,58 @@ describe('PSocket connect', () => { expect(client.connect).toHaveBeenCalledWith('example.com', 443); }); - it('retries with a refreshed client when the first attempt fails', async () => { + it('retries once when the client could not be built', async () => { const client = makeClient(makeStream()); mockGetEpoxyClient - .mockRejectedValueOnce(new Error('stale client')) + .mockRejectedValueOnce(new Error('wasm unavailable')) .mockResolvedValueOnce(client); const socket = new PSocket('example.com', 80); const events = listen(socket); await until(() => expect(events.open).toHaveBeenCalledTimes(1)); - expect(mockGetEpoxyClient).toHaveBeenNthCalledWith(1, { refresh: false }); - expect(mockGetEpoxyClient).toHaveBeenNthCalledWith(2, { refresh: true }); + expect(mockGetEpoxyClient).toHaveBeenCalledTimes(2); expect(events.error).not.toHaveBeenCalled(); }); + it('retries on a fresh client when the relay itself failed', async () => { + const relayError = new Error('Wisp: Multiplexor task ended'); + const dead = makeClient(makeStream()); + dead.connect.mockRejectedValue(relayError); + const fresh = makeClient(makeStream()); + mockGetEpoxyClient + .mockResolvedValueOnce(dead) + .mockResolvedValueOnce(fresh); + mockDiscardOnRelayFailure.mockReturnValueOnce(true); + + const socket = new PSocket('example.com', 80); + const events = listen(socket); + await until(() => expect(events.open).toHaveBeenCalledTimes(1)); + + expect(mockDiscardOnRelayFailure).toHaveBeenCalledWith( + dead, + relayError, + ); + expect(fresh.connect).toHaveBeenCalledWith('example.com', 80); + expect(events.error).not.toHaveBeenCalled(); + }); + + // One unreachable host must not cost a second relay client. + it('does not retry a stream failure on a working relay', async () => { + const refused = new Error('connection refused'); + const client = makeClient(makeStream()); + client.connect.mockRejectedValue(refused); + mockGetEpoxyClient.mockResolvedValue(client); + + const socket = new PSocket('down.example', 80); + const events = listen(socket); + await until(() => expect(events.close).toHaveBeenCalledWith(true)); + + expect(mockGetEpoxyClient).toHaveBeenCalledTimes(1); + expect(mockDiscardOnRelayFailure).toHaveBeenCalledWith(client, refused); + expect(events.error.mock.calls[0][0]).toBe(refused); + }); + it('emits an error and closes when both attempts fail', async () => { mockGetEpoxyClient.mockRejectedValue(new Error('relay down')); @@ -185,7 +222,7 @@ describe('PSocket connect', () => { expect(events.open).not.toHaveBeenCalled(); expect(events.error).toHaveBeenCalledTimes(1); expect(events.close).toHaveBeenCalledWith(true); - expect(mockClearEpoxyClientCache).toHaveBeenCalled(); + expect(mockGetEpoxyClient).toHaveBeenCalledTimes(2); }); // The types declare the handler as `(error: Error) => void` and document the @@ -246,14 +283,15 @@ describe('PSocket inbound data', () => { expect(events.error).not.toHaveBeenCalled(); }); - it('emits an error and drops the cached client when the read fails', async () => { - const { events, stream } = await connected(); + it('emits an error and reports it to the client cache when the read fails', async () => { + const { events, stream, client } = await connected(); - stream.failRead(new Error('connection reset')); + const reset = new Error('connection reset'); + stream.failRead(reset); await until(() => expect(events.error).toHaveBeenCalled()); expect(events.error.mock.calls[0][0].message).toBe('connection reset'); - expect(mockClearEpoxyClientCache).toHaveBeenCalled(); + expect(mockDiscardOnRelayFailure).toHaveBeenCalledWith(client, reset); await until(() => expect(events.close).toHaveBeenCalledWith(true)); }); }); @@ -311,6 +349,71 @@ describe('PSocket write', () => { await until(() => expect(callback).toHaveBeenCalledTimes(1)); }); + it('sends queued writes before writes from the open handler', async () => { + const stream = makeStream(); + const gate = deferred(); + mockGetEpoxyClient.mockReturnValue(gate.promise); + const socket = new PSocket('example.com', 80); + socket.write('first'); + socket.on('open', () => socket.write('second')); + gate.resolve(makeClient(stream)); + await until(() => expect(stream.written).toHaveLength(2)); + expect( + stream.written.map((chunk) => new TextDecoder().decode(chunk)), + ).toEqual(['first', 'second']); + socket.close(); + }); + + it('queues a write made during the flush behind the queued writes', async () => { + const written = []; + const sinkGate = deferred(); + let sinkCalls = 0; + const stream = { + read: new ReadableStream(), + write: new WritableStream({ + async write(chunk) { + written.push(new TextDecoder().decode(chunk)); + if (sinkCalls++ === 0) await sinkGate.promise; + }, + }), + }; + const gate = deferred(); + mockGetEpoxyClient.mockReturnValue(gate.promise); + const socket = new PSocket('example.com', 80); + socket.write('first'); + socket.write('second'); + gate.resolve(makeClient(stream)); + + // The flush is parked on 'first'; this write comes from elsewhere. + await until(() => expect(written).toEqual(['first'])); + socket.write('third'); + sinkGate.resolve(); + + await until(() => expect(written).toHaveLength(3)); + expect(written).toEqual(['first', 'second', 'third']); + socket.close(); + }); + + it('sends a write made right after the flush drains the queue', async () => { + const stream = makeStream(); + const gate = deferred(); + mockGetEpoxyClient.mockReturnValue(gate.promise); + const socket = new PSocket('example.com', 80); + const events = listen(socket); + const callback = vi.fn(); + + // Two microtasks out lands after the flush loop's last queue check but + // before the connect path resumes. + const writeLate = () => queueMicrotask(() => queueMicrotask(() => socket.write('second', callback))); + socket.write('first', writeLate); + gate.resolve(makeClient(stream)); + + await until(() => expect(events.open).toHaveBeenCalled()); + await until(() => expect(stream.written).toHaveLength(2)); + await until(() => expect(callback).toHaveBeenCalledTimes(1)); + socket.close(); + }); + it('queues writes issued before the socket opens and flushes them in order', async () => { const stream = makeStream(); const gate = deferred(); @@ -342,15 +445,19 @@ describe('PSocket write', () => { expect(() => socket.write('late')).toThrow(/already closed/); }); - it('emits an error and drops the cached client when a write fails', async () => { - const stream = makeStream({ failWrite: new Error('write failed') }); - const { socket, events } = await connected(stream); + it('emits an error and reports it to the client cache when a write fails', async () => { + const failWrite = new Error('write failed'); + const stream = makeStream({ failWrite }); + const { socket, events, client } = await connected(stream); socket.write('doomed'); await until(() => expect(events.error).toHaveBeenCalled()); expect(events.error.mock.calls[0][0].message).toBe('write failed'); - expect(mockClearEpoxyClientCache).toHaveBeenCalled(); + expect(mockDiscardOnRelayFailure).toHaveBeenCalledWith( + client, + failWrite, + ); }); // Regression guard: a failed write hands the close event to #closeStreams, @@ -387,6 +494,63 @@ describe('PSocket write', () => { }); }); +describe('PSocket throwing listeners', () => { + // Listener throws are rethrown on a 0 ms timer; capture those rather + // than letting them fail the run. + const captureRethrows = () => { + const rethrows = []; + const realSetTimeout = globalThis.setTimeout; + vi.spyOn(globalThis, 'setTimeout').mockImplementation((fn, ms, ...args) => { + if ( ms === 0 ) { + rethrows.push(fn); + return 0; + } + return realSetTimeout(fn, ms, ...args); + }); + return rethrows; + }; + + it('closes once when an error listener throws', async () => { + const rethrows = captureRethrows(); + const stream = makeStream({ failWrite: new Error('write failed') }); + const { socket, events } = await connected(stream); + const listenerError = new Error('handler bug'); + socket.on('error', () => { + throw listenerError; + }); + + socket.write('doomed'); + await until(() => expect(events.close).toHaveBeenCalled()); + + expect(events.error).toHaveBeenCalledTimes(1); + expect(events.close).toHaveBeenCalledTimes(1); + expect(events.close).toHaveBeenCalledWith(true); + expect(rethrows).toHaveLength(1); + expect(() => rethrows[0]()).toThrow(listenerError); + }); + + it('does not reconnect when an open listener throws', async () => { + const rethrows = captureRethrows(); + const stream = makeStream(); + mockGetEpoxyClient.mockResolvedValue(makeClient(stream)); + mockDiscardOnRelayFailure.mockReturnValue(true); + const socket = new PSocket('example.com', 80); + const events = listen(socket); + socket.on('open', () => { + throw new Error('handler bug'); + }); + + await until(() => expect(rethrows).toHaveLength(1)); + socket.write('still works'); + await until(() => expect(stream.written).toHaveLength(1)); + + expect(events.open).toHaveBeenCalledTimes(1); + expect(mockGetEpoxyClient).toHaveBeenCalledTimes(1); + expect(events.error).not.toHaveBeenCalled(); + socket.close(); + }); +}); + describe('PSocket close', () => { it('emits close exactly once, even when called repeatedly', async () => { const { socket, events } = await connected(); diff --git a/src/puter-js/src/modules/networking/epoxy.js b/src/puter-js/src/modules/networking/epoxy.js index 952397533..d31c0dddd 100644 --- a/src/puter-js/src/modules/networking/epoxy.js +++ b/src/puter-js/src/modules/networking/epoxy.js @@ -76,11 +76,74 @@ export function createPuterPasswordBuilder (runtime, wispToken) { return new PuterPasswordExtBuilder(['', wispToken]); } -export let initEpoxy = async ({ wispToken, wispServer }) => { +/** + * The runtime's `WebSocketJsProvider`, except that each relay websocket is + * recorded on `transport`: `down` says whether the current one is unusable, and + * a `closed` transport refuses to dial again. + * + * @param {string} url + * @param {string | undefined} protocol + * @param {import('./types.js').RelayTransport} transport + * @returns {Promise<[ReadableStream, WritableStream]>} + */ +export function openRelaySocket (url, protocol, transport) { + if ( transport.closed ) { + return Promise.reject(new Error('Relay connection was closed.')); + } + + const ws = new WebSocket(url, protocol ? [protocol] : []); + ws.binaryType = 'arraybuffer'; + transport.sockets.add(ws); + transport.current = ws; + + const markDown = () => { + transport.sockets.delete(ws); + if ( transport.current === ws ) { + transport.down = true; + } + }; + ws.addEventListener('error', markDown); + ws.addEventListener('close', markDown); + + return new Promise((resolve, reject) => { + ws.addEventListener('error', reject); + ws.addEventListener('open', () => { + ws.removeEventListener('error', reject); + if ( transport.current === ws ) { + transport.down = false; + } + + const readable = new ReadableStream({ + start (controller) { + ws.addEventListener('message', ({ data }) => { + controller.enqueue(typeof data === 'string' ? data : new Uint8Array(data)); + }); + ws.addEventListener('error', error => controller.error(error)); + ws.addEventListener('close', () => { + try { + controller.close(); + } catch { + // already errored + } + }); + }, + cancel: () => ws.close(), + }); + const writable = new WritableStream({ + write: chunk => ws.send(chunk), + abort: () => ws.close(), + close: () => ws.close(), + }); + resolve([readable, writable]); + }); + }); +} + +export let initEpoxy = async ({ wispToken, wispServer, transport }) => { const runtime = await getEpoxyRuntime(); const provider = new runtime.WispSocketProvider( - new runtime.WebSocketJsProvider(), + new runtime.JsProvider((url, protocol) => openRelaySocket(url, protocol, transport)), wispServer, () => ({ builders: [createPuterPasswordBuilder(runtime, wispToken)], diff --git a/src/puter-js/src/modules/networking/epoxy.test.js b/src/puter-js/src/modules/networking/epoxy.test.js index 0f4673257..0aad0b1db 100644 --- a/src/puter-js/src/modules/networking/epoxy.test.js +++ b/src/puter-js/src/modules/networking/epoxy.test.js @@ -1,5 +1,5 @@ -import { describe, expect, it } from 'vitest'; -import { createPuterPasswordBuilder } from './epoxy.js'; +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; +import { createPuterPasswordBuilder, openRelaySocket } from './epoxy.js'; // Stand-ins for the extension base classes the epoxy wasm bundle exports. // Loading the real runtime needs a network fetch plus a wasm instantiation, so @@ -77,3 +77,150 @@ describe('puter password extension', () => { expect(build().buildFromBytes(new Uint8Array([0])).required).toBe(false); }); }); + +describe('openRelaySocket', () => { + class FakeWebSocket extends EventTarget { + static instances = []; + + constructor(url, protocols) { + super(); + this.url = url; + this.protocols = protocols; + this.send = vi.fn(); + this.close = vi.fn(() => this.fire('close')); + FakeWebSocket.instances.push(this); + } + + fire(type, props = {}) { + this.dispatchEvent(Object.assign(new Event(type), props)); + } + } + + const origWebSocket = globalThis.WebSocket; + const newTransport = () => ({ + sockets: new Set(), + current: undefined, + down: false, + closed: false, + }); + + beforeEach(() => { + FakeWebSocket.instances = []; + globalThis.WebSocket = FakeWebSocket; + }); + + afterEach(() => { + globalThis.WebSocket = origWebSocket; + }); + + it('hands epoxy a readable and writable over the opened websocket', async () => { + const transport = newTransport(); + const opened = openRelaySocket( + 'wss://relay.test/', + 'wisp-v2', + transport, + ); + const [ws] = FakeWebSocket.instances; + expect(ws.url).toBe('wss://relay.test/'); + expect(ws.protocols).toEqual(['wisp-v2']); + expect(ws.binaryType).toBe('arraybuffer'); + ws.fire('open'); + const [readable, writable] = await opened; + + ws.fire('message', { data: new Uint8Array([1, 2]).buffer }); + ws.fire('message', { data: 'text' }); + const reader = readable.getReader(); + expect(Array.from((await reader.read()).value)).toEqual([1, 2]); + expect((await reader.read()).value).toBe('text'); + + const chunk = new Uint8Array([3]); + await writable.getWriter().write(chunk); + expect(ws.send).toHaveBeenCalledWith(chunk); + expect(transport.sockets.has(ws)).toBe(true); + expect(transport.down).toBe(false); + }); + + it('marks the transport down when the relay closes the websocket', async () => { + const transport = newTransport(); + const opened = openRelaySocket( + 'wss://relay.test/', + undefined, + transport, + ); + const [ws] = FakeWebSocket.instances; + ws.fire('open'); + const [readable] = await opened; + + ws.fire('close'); + + expect(transport.down).toBe(true); + expect(transport.sockets.size).toBe(0); + expect((await readable.getReader().read()).done).toBe(true); + }); + + it('rejects and marks the transport down when the websocket cannot open', async () => { + const transport = newTransport(); + const opened = openRelaySocket( + 'wss://relay.test/', + undefined, + transport, + ); + FakeWebSocket.instances[0].fire('error'); + + await expect(opened).rejects.toBeInstanceOf(Event); + expect(transport.down).toBe(true); + }); + + it('clears the down flag once a redial opens', async () => { + const transport = newTransport(); + const first = openRelaySocket( + 'wss://relay.test/', + undefined, + transport, + ); + FakeWebSocket.instances[0].fire('open'); + await first; + FakeWebSocket.instances[0].fire('close'); + + const second = openRelaySocket( + 'wss://relay.test/', + undefined, + transport, + ); + FakeWebSocket.instances[1].fire('open'); + await second; + + expect(transport.down).toBe(false); + }); + + it('ignores a superseded websocket closing late', async () => { + const transport = newTransport(); + const first = openRelaySocket( + 'wss://relay.test/', + undefined, + transport, + ); + FakeWebSocket.instances[0].fire('open'); + await first; + const second = openRelaySocket( + 'wss://relay.test/', + undefined, + transport, + ); + FakeWebSocket.instances[1].fire('open'); + await second; + + FakeWebSocket.instances[0].fire('close'); + + expect(transport.down).toBe(false); + }); + + it('refuses to dial once the transport is closed', async () => { + const transport = { ...newTransport(), closed: true }; + + await expect( + openRelaySocket('wss://relay.test/', undefined, transport), + ).rejects.toThrow(/closed/); + expect(FakeWebSocket.instances).toHaveLength(0); + }); +}); diff --git a/src/puter-js/src/modules/networking/index.js b/src/puter-js/src/modules/networking/index.js index 840a5c354..681cac0ea 100644 --- a/src/puter-js/src/modules/networking/index.js +++ b/src/puter-js/src/modules/networking/index.js @@ -3,6 +3,18 @@ import { PSocket, PTLSSocket } from './PSocket.js'; import { pFetch } from './requests.js'; let cachedEpoxy = undefined; +/** @type {WeakMap} */ +const epoxyEntries = new WeakMap(); + +// What wisp reports when the relay connection itself is dead or refused us, as +// opposed to one stream failing (refused port, DNS, TLS, HTTP). +const RELAY_FAILURE_PATTERN = new RegExp([ + 'multiplexor', + 'websocket implementation error', + 'handshake timed out', + 'incompatible wisp protocol version', + 'protocol extension', +].join('|'), 'i'); function getPuterInstance () { const puter = globalThis.puter; @@ -76,21 +88,23 @@ export async function generateWispV1URL () { return `${wispServer}/${wispToken}/`; } -export async function getEpoxyClient ({ refresh = false } = {}) { +export async function getEpoxyClient () { const nextKey = getClientCacheKey(); - // Concurrent callers share one attempt, but only while it is still the - // attempt they asked for: a `refresh` exists to replace the cached entry, - // and a changed origin or token needs a client of its own. - if ( ! refresh && cachedEpoxy && cachedEpoxy.key === nextKey ) { + // Concurrent callers share one attempt; a changed origin or token needs a + // client of its own. + if ( cachedEpoxy && cachedEpoxy.key === nextKey ) { return await cachedEpoxy.promise; } - const epoxy = { key: nextKey }; + const transport = { sockets: new Set(), current: undefined, down: false, closed: false }; + const epoxy = { key: nextKey, transport }; epoxy.promise = (async () => { try { const { wispToken, wispServer } = await getWispCredentials(); - return await initEpoxy({ wispToken, wispServer }); + const client = await initEpoxy({ wispToken, wispServer, transport }); + epoxyEntries.set(client, epoxy); + return client; } catch ( error ) { // Never cache a failed attempt, or every later caller would keep // resolving the same broken client. The reason is re-thrown so @@ -112,6 +126,43 @@ export function clearEpoxyClientCache () { cachedEpoxy = undefined; } +/** + * Drops `client` from the cache and closes its relay websocket, but only when + * `error` means the relay connection is unusable. A failure of one stream + * leaves the shared client alone. + * + * @param {object | undefined} client + * @param {unknown} error + * @returns {boolean} whether it was a relay failure + */ +export function discardOnRelayFailure (client, error) { + const entry = client && epoxyEntries.get(client); + if ( ! entry ) { + return false; + } + + const { transport } = entry; + const message = error instanceof Error ? error.message : String(error); + if ( !transport.down && !RELAY_FAILURE_PATTERN.test(message) ) { + return false; + } + + if ( cachedEpoxy === entry ) { + cachedEpoxy = undefined; + } + transport.closed = true; + transport.down = true; + for ( const ws of transport.sockets ) { + try { + ws.close(); + } catch { + // ignored + } + } + transport.sockets.clear(); + return true; +} + /** * The `puter.net` module. * diff --git a/src/puter-js/src/modules/networking/index.test.js b/src/puter-js/src/modules/networking/index.test.js index c0d275ee7..e2d128f9d 100644 --- a/src/puter-js/src/modules/networking/index.test.js +++ b/src/puter-js/src/modules/networking/index.test.js @@ -9,6 +9,7 @@ vi.mock('./epoxy.js', () => ({ const { clearEpoxyClientCache, + discardOnRelayFailure, generateWispV1URL, getEpoxyClient, getWispCredentials, @@ -184,6 +185,7 @@ describe('getEpoxyClient', () => { expect(mockInitEpoxy).toHaveBeenCalledWith({ wispToken: CREDENTIALS.token, wispServer: CREDENTIALS.server, + transport: expect.objectContaining({ down: false, closed: false }), }); }); @@ -212,13 +214,6 @@ describe('getEpoxyClient', () => { expect(mockInitEpoxy).toHaveBeenCalledTimes(2); }); - it('rebuilds the client when a refresh is requested', async () => { - await getEpoxyClient(); - await getEpoxyClient({ refresh: true }); - - expect(mockInitEpoxy).toHaveBeenCalledTimes(2); - }); - it('shares one in-flight init between concurrent callers', async () => { const gate = deferred(); mockInitEpoxy.mockImplementation(() => gate.promise); @@ -283,21 +278,6 @@ describe('getEpoxyClient', () => { expect(retried).toEqual({ client: 'epoxy' }); }); - // A refresh exists to replace what is cached, so it must not be answered - // with the very attempt the caller is trying to supersede. - it('honours a refresh requested while an init is still in flight', async () => { - const gate = deferred(); - mockInitEpoxy.mockImplementationOnce(() => gate.promise); - - const stale = getEpoxyClient(); - const fresh = getEpoxyClient({ refresh: true }); - gate.resolve({ client: 'stale' }); - - expect(await stale).toEqual({ client: 'stale' }); - expect(await fresh).toEqual({ client: 'epoxy' }); - expect(mockInitEpoxy).toHaveBeenCalledTimes(2); - }); - it('starts a separate attempt when the token changes mid-init', async () => { const gate = deferred(); mockInitEpoxy.mockImplementationOnce(() => gate.promise); @@ -313,6 +293,101 @@ describe('getEpoxyClient', () => { }); }); +describe('discardOnRelayFailure', () => { + const transports = new Map(); + const transportOf = (client) => transports.get(client); + + beforeEach(() => { + transports.clear(); + mockInitEpoxy.mockImplementation(async ({ transport }) => { + const client = { client: 'epoxy' }; + transports.set(client, transport); + return client; + }); + }); + + it.each([ + 'connection refused', + 'IO: tls handshake eof', + 'Invalid DNS name: down.invalid', + 'Hyper: connection closed before message completed', + 'TLS: invalid peer certificate: UnknownIssuer', + ])( + 'keeps the shared client after a stream failure (%s)', + async (message) => { + const client = await getEpoxyClient(); + + expect(discardOnRelayFailure(client, new Error(message))).toBe( + false, + ); + expect(await getEpoxyClient()).toBe(client); + expect(mockFetch).toHaveBeenCalledTimes(1); + }, + ); + + it.each([ + 'Wisp: Multiplexor task ended', + 'IO: Failed to send multiplexor message', + 'Wisp: Websocket implementation error: websocket closed', + 'Wisp: Handshake timed out', + 'Wisp: Password protocol extension: Invalid username/password', + 'Wisp: Incompatible Wisp protocol version: found 1.0 but needed 2.0', + ])('drops the client on a relay failure (%s)', async (message) => { + const client = await getEpoxyClient(); + + expect(discardOnRelayFailure(client, new Error(message))).toBe(true); + expect(await getEpoxyClient()).not.toBe(client); + expect(mockFetch).toHaveBeenCalledTimes(2); + }); + + it('treats any failure as a relay failure once the relay websocket is down', async () => { + const client = await getEpoxyClient(); + transportOf(client).down = true; + + expect( + discardOnRelayFailure(client, new Error('connection reset')), + ).toBe(true); + }); + + it("closes the dropped client's websockets and stops it redialling", async () => { + const client = await getEpoxyClient(); + const transport = transportOf(client); + const ws = { close: vi.fn() }; + transport.sockets.add(ws); + + discardOnRelayFailure( + client, + new Error('Wisp: Multiplexor task ended'), + ); + + expect(ws.close).toHaveBeenCalledTimes(1); + expect(transport.closed).toBe(true); + expect(transport.sockets.size).toBe(0); + }); + + it('leaves a newer cached client alone when a stale one fails', async () => { + const stale = await getEpoxyClient(); + discardOnRelayFailure(stale, new Error('Wisp: Multiplexor task ended')); + const fresh = await getEpoxyClient(); + + expect( + discardOnRelayFailure( + stale, + new Error('Wisp: Multiplexor task ended'), + ), + ).toBe(true); + expect(await getEpoxyClient()).toBe(fresh); + expect(transportOf(fresh).closed).toBe(false); + }); + + it('ignores a missing or unknown client', () => { + const error = new Error('Wisp: Multiplexor task ended'); + + expect(discardOnRelayFailure(undefined, error)).toBe(false); + expect(discardOnRelayFailure({}, error)).toBe(false); + }); +}); + describe('netAPI surface', () => { it('exposes the socket constructors and fetch the docs promise', () => { expect(typeof netAPI.Socket).toBe('function'); @@ -351,6 +426,51 @@ describe('failure reporting through a real socket', () => { expect(events.close).toHaveBeenCalledWith(true); }); + it('reports a refused port without building a second client', async () => { + mockInitEpoxy.mockImplementation(async () => ({ + connect: async () => { + throw new Error('connection refused'); + }, + })); + + const first = listen(); + await closed(first); + const second = listen(); + await closed(second); + + expect(first.error.mock.calls[0][0].message).toBe('connection refused'); + expect(mockInitEpoxy).toHaveBeenCalledTimes(1); + expect(mockFetch).toHaveBeenCalledTimes(1); + }); + + it('moves to a new client when the relay connection died', async () => { + const stream = { + read: new ReadableStream(), + write: new WritableStream(), + }; + const dead = { + connect: vi.fn(async () => { + throw new Error('Wisp: Multiplexor task ended'); + }), + }; + const fresh = { connect: vi.fn(async () => stream) }; + mockInitEpoxy.mockResolvedValueOnce(dead).mockResolvedValueOnce(fresh); + + const socket = new netAPI.Socket('example.com', 80); + const open = vi.fn(); + const error = vi.fn(); + socket.on('open', open); + socket.on('error', error); + await vi.waitFor(() => expect(open).toHaveBeenCalled(), { + interval: 1, + }); + + expect(fresh.connect).toHaveBeenCalledWith('example.com', 80); + expect(error).not.toHaveBeenCalled(); + expect(mockFetch).toHaveBeenCalledTimes(2); + socket.close(); + }); + it('reports why the relay refused to mint a token', async () => { mockFetch.mockImplementation(async () => relayResponse({ status: 503 })); diff --git a/src/puter-js/src/modules/networking/requests.js b/src/puter-js/src/modules/networking/requests.js index 8542d84d6..2f8005dcb 100644 --- a/src/puter-js/src/modules/networking/requests.js +++ b/src/puter-js/src/modules/networking/requests.js @@ -1,4 +1,4 @@ -import { clearEpoxyClientCache, getEpoxyClient } from './index.js'; +import { discardOnRelayFailure, getEpoxyClient } from './index.js'; function logFetchResult ({ params, result, error }) { if ( ! globalThis.puter?.apiCallLogger?.isEnabled() ) { @@ -94,13 +94,12 @@ async function assertRequestIsFetchable (args) { */ export async function pFetch (...args) { const params = getFetchLogParams(args); - let usedEpoxyClient = false; + let client; try { await assertRequestIsFetchable(args); - const client = await getEpoxyClient(); - usedEpoxyClient = true; + client = await getEpoxyClient(); const response = await client.fetch(...args); logFetchResult({ @@ -113,9 +112,7 @@ export async function pFetch (...args) { return response; } catch ( error ) { - if ( usedEpoxyClient ) { - clearEpoxyClientCache(); - } + discardOnRelayFailure(client, error); logFetchResult({ params, diff --git a/src/puter-js/src/modules/networking/requests.test.js b/src/puter-js/src/modules/networking/requests.test.js index e2fc60598..469515a4c 100644 --- a/src/puter-js/src/modules/networking/requests.test.js +++ b/src/puter-js/src/modules/networking/requests.test.js @@ -3,10 +3,10 @@ import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; // pFetch is a thin wrapper over the epoxy client, so the client module is // stubbed: the tests care about delegation, cache invalidation, and logging. const mockGetEpoxyClient = vi.fn(); -const mockClearEpoxyClientCache = vi.fn(); +const mockDiscardOnRelayFailure = vi.fn(); vi.mock('./index.js', () => ({ getEpoxyClient: (...args) => mockGetEpoxyClient(...args), - clearEpoxyClientCache: (...args) => mockClearEpoxyClientCache(...args), + discardOnRelayFailure: (...args) => mockDiscardOnRelayFailure(...args), })); const { pFetch } = await import('./requests.js'); @@ -14,16 +14,17 @@ const { pFetch } = await import('./requests.js'); const origPuter = globalThis.puter; let mockClientFetch; +let client; let logRequest; // Matches what the epoxy client resolves to: a Response-like object. const RESPONSE = { status: 204, statusText: 'No Content' }; beforeEach(() => { - mockClearEpoxyClientCache.mockReset(); + mockDiscardOnRelayFailure.mockReset().mockReturnValue(false); mockClientFetch = vi.fn(async () => RESPONSE); - mockGetEpoxyClient.mockReset() - .mockResolvedValue({ fetch: (...args) => mockClientFetch(...args) }); + client = { fetch: (...args) => mockClientFetch(...args) }; + mockGetEpoxyClient.mockReset().mockResolvedValue(client); logRequest = vi.fn(); globalThis.puter = { @@ -57,22 +58,36 @@ describe('pFetch delegation', () => { await expect(pFetch('https://example.com')).rejects.toThrow('socket closed'); }); - it('drops the cached client when the request fails', async () => { - mockClientFetch.mockRejectedValue(new Error('socket closed')); + it('reports a failed request to the client cache', async () => { + const error = new Error('socket closed'); + mockClientFetch.mockRejectedValue(error); - await expect(pFetch('https://example.com')).rejects.toThrow(); + await expect(pFetch('https://example.com')).rejects.toThrow( + 'socket closed', + ); - expect(mockClearEpoxyClientCache).toHaveBeenCalledTimes(1); + expect(mockDiscardOnRelayFailure).toHaveBeenCalledWith(client, error); }); - // Nothing was cached if the client itself never came up, so there is - // nothing to invalidate -- clearing here would just mask the real failure. - it('leaves the cache alone when the client could not be created', async () => { - mockGetEpoxyClient.mockRejectedValue(new Error('wasm unavailable')); + it('rethrows unchanged when the failure was a relay failure', async () => { + const error = new Error('Wisp: Multiplexor task ended'); + mockClientFetch.mockRejectedValue(error); + mockDiscardOnRelayFailure.mockReturnValue(true); + + await expect(pFetch('https://example.com')).rejects.toBe(error); + }); + + // No client came up, so there is nothing to discard. + it('reports no client when the client could not be created', async () => { + const error = new Error('wasm unavailable'); + mockGetEpoxyClient.mockRejectedValue(error); await expect(pFetch('https://example.com')).rejects.toThrow('wasm unavailable'); - expect(mockClearEpoxyClientCache).not.toHaveBeenCalled(); + expect(mockDiscardOnRelayFailure).toHaveBeenCalledWith( + undefined, + error, + ); }); }); @@ -225,13 +240,17 @@ describe('pFetch request validation', () => { await expect(request.text()).resolves.toBe('payload'); }); - it('logs a validation failure without clearing the client cache', async () => { + it('logs a validation failure without dialing', async () => { await expect(pFetch('ftp://example.com/file.txt')).rejects.toBeTruthy(); const [entry] = logRequest.mock.calls[0]; expect(entry.error.message).toBe( 'Failed to fetch. URL scheme "ftp:" is not supported.', ); - expect(mockClearEpoxyClientCache).not.toHaveBeenCalled(); + expect(mockGetEpoxyClient).not.toHaveBeenCalled(); + expect(mockDiscardOnRelayFailure).toHaveBeenCalledWith( + undefined, + expect.anything(), + ); }); }); diff --git a/src/puter-js/src/modules/networking/types.js b/src/puter-js/src/modules/networking/types.js index 7c6abab1d..19a148c6c 100644 --- a/src/puter-js/src/modules/networking/types.js +++ b/src/puter-js/src/modules/networking/types.js @@ -29,4 +29,16 @@ * Fetch an http/https resource without being bound by CORS restrictions. */ +/** + * The relay websockets behind one epoxy client. + * + * @typedef {{ + * sockets: Set; + * current: WebSocket | undefined; + * down: boolean; + * closed: boolean; + * }} RelayTransport + * @internal + */ + export {};