diff --git a/src/puter-js/src/modules/networking/PSocket.js b/src/puter-js/src/modules/networking/PSocket.js index b78b78986..da511fde4 100644 --- a/src/puter-js/src/modules/networking/PSocket.js +++ b/src/puter-js/src/modules/networking/PSocket.js @@ -178,14 +178,21 @@ export class PSocket extends EventListener { } } - this.#emitClose(false); + // A teardown already under way owns the close event. Cancelling the + // reader resolves the pending read with `done`, which lands here + // first; emitting now would report a clean shutdown and let + // #closeStreams' hadError flag lose the race. + if ( ! this.#closing ) { + this.#emitClose(false); + } } catch ( error ) { if ( this.#closing ) { - this.#emitClose(false); - } else { - clearEpoxyClientCache(); - this.#emitErrorAndClose(error); + // As above: #closeStreams is mid-flight and will emit close. + return; } + + clearEpoxyClientCache(); + this.#emitErrorAndClose(error); } finally { try { this.#reader.releaseLock(); diff --git a/src/puter-js/src/modules/networking/PSocket.test.js b/src/puter-js/src/modules/networking/PSocket.test.js index 11ed1c149..cf4bc86cb 100644 --- a/src/puter-js/src/modules/networking/PSocket.test.js +++ b/src/puter-js/src/modules/networking/PSocket.test.js @@ -50,6 +50,41 @@ function makeStream ({ failWrite } = {}) { }; } +// The epoxy client's streams come from wasm rather than being the platform's +// ReadableStream, so a cancel that rejects the pending read -- instead of +// resolving it `done` the way the spec requires -- is possible. This models +// that, which a real ReadableStream cannot be made to do. +function makeRejectingStream ({ failWrite } = {}) { + let rejectRead; + const reader = { + read: () => new Promise((resolve, reject) => { + rejectRead = reject; + }), + cancel: async () => { + rejectRead?.(new Error('stream torn down')); + }, + releaseLock: () => {}, + }; + + const written = []; + const writer = { + write: async chunk => { + if ( failWrite ) { + throw failWrite; + } + written.push(chunk); + }, + close: async () => {}, + releaseLock: () => {}, + }; + + return { + read: { getReader: () => reader }, + write: { getWriter: () => writer }, + written, + }; +} + function makeClient (stream) { return { connect: vi.fn(async () => stream), @@ -318,17 +353,11 @@ describe('PSocket write', () => { expect(mockClearEpoxyClientCache).toHaveBeenCalled(); }); - // KNOWN BUG -- unskip once #readLoop stops racing the error path. - // - // A failed write sets #closing and hands the close event to #closeStreams, - // which must await reader.cancel() and writer.close() before emitting - // close(true). That cancel resolves #readLoop's pending read() with - // {done: true}, so the loop breaks and reaches its own #emitClose(false) - // first -- #closed is already set by the time #closeStreams gets there, so - // callers are told the socket shut down cleanly after an error. - // A read failure is unaffected: that path throws into #readLoop's catch, - // which never calls #emitClose(false). - it.skip('closes with the error flag set when a write fails', async () => { + // Regression guard: a failed write hands the close event to #closeStreams, + // whose reader.cancel() resolves #readLoop's pending read with {done: true}. + // The loop must not treat that as a clean shutdown and emit close(false) + // before #closeStreams reports the error. + it('closes with the error flag set when a write fails', async () => { const stream = makeStream({ failWrite: new Error('write failed') }); const { socket, events } = await connected(stream); @@ -337,6 +366,25 @@ describe('PSocket write', () => { expect(events.close).toHaveBeenCalledWith(true); }); + + // Same guarantee when the teardown's cancel rejects the in-flight read + // rather than ending it cleanly: the flag must still come from the path + // that knows an error happened. + it('keeps the error flag when cancelling rejects the pending read', async () => { + const stream = makeRejectingStream({ failWrite: new Error('write failed') }); + mockGetEpoxyClient.mockResolvedValue(makeClient(stream)); + + const socket = new PSocket('example.com', 80); + const events = listen(socket); + await until(() => expect(events.open).toHaveBeenCalled()); + + socket.write('doomed'); + await until(() => expect(events.close).toHaveBeenCalled()); + + expect(events.error.mock.calls[0][0].message).toBe('write failed'); + expect(events.close).toHaveBeenCalledTimes(1); + expect(events.close).toHaveBeenCalledWith(true); + }); }); describe('PSocket close', () => { @@ -372,6 +420,34 @@ describe('PSocket close', () => { expect(events.open).not.toHaveBeenCalled(); }); + // #readLoop now defers the close event to #closeStreams whenever a teardown + // is under way, so #closeStreams has to emit it even when cancelling the + // reader fails -- otherwise close would go missing entirely. + it('still emits close when cancelling the reader fails', async () => { + let readController; + const read = new ReadableStream({ + start (controller) { + readController = controller; + }, + cancel () { + throw new Error('cancel failed'); + }, + }); + void readController; + const write = new WritableStream({ write () {} }); + mockGetEpoxyClient.mockResolvedValue(makeClient({ read, write })); + + const socket = new PSocket('example.com', 80); + const events = listen(socket); + await until(() => expect(events.open).toHaveBeenCalled()); + + socket.close(); + await until(() => expect(events.close).toHaveBeenCalled()); + + expect(events.close).toHaveBeenCalledTimes(1); + expect(events.close).toHaveBeenCalledWith(false); + }); + it('stops emitting data after close', async () => { const { socket, events, stream } = await connected(); diff --git a/src/puter-js/src/modules/networking/index.js b/src/puter-js/src/modules/networking/index.js index 8a0bca94c..b369c414f 100644 --- a/src/puter-js/src/modules/networking/index.js +++ b/src/puter-js/src/modules/networking/index.js @@ -77,29 +77,35 @@ export async function generateWispV1URL () { } export async function getEpoxyClient ({ refresh = false } = {}) { - if ( cachedEpoxy && cachedEpoxy.initting ) return await cachedEpoxy.promise; - const nextKey = getClientCacheKey(); - if ( refresh || !(cachedEpoxy && cachedEpoxy.key === nextKey) ) { - let epoxy = { key: nextKey, initting: true }; - let promise = (async () => { - try { - const { wispToken, wispServer } = await getWispCredentials(); - let ret = await initEpoxy({ wispToken, wispServer }); - epoxy.initting = false; - return ret; - } catch { - if ( cachedEpoxy === epoxy ) { - cachedEpoxy = undefined; - } - } - })(); - epoxy.promise = promise; - cachedEpoxy = epoxy; + // 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 ) { + return await cachedEpoxy.promise; } - return await cachedEpoxy.promise; + const epoxy = { key: nextKey }; + epoxy.promise = (async () => { + try { + const { wispToken, wispServer } = await getWispCredentials(); + return await initEpoxy({ wispToken, wispServer }); + } catch ( error ) { + // Never cache a failed attempt, or every later caller would keep + // resolving the same broken client. The reason is re-thrown so + // callers report why the relay is unreachable instead of tripping + // over an undefined client. + if ( cachedEpoxy === epoxy ) { + cachedEpoxy = undefined; + } + throw error; + } + })(); + + cachedEpoxy = epoxy; + + return await epoxy.promise; } export function clearEpoxyClientCache () { diff --git a/src/puter-js/src/modules/networking/index.test.js b/src/puter-js/src/modules/networking/index.test.js index 5a3ae265f..c0d275ee7 100644 --- a/src/puter-js/src/modules/networking/index.test.js +++ b/src/puter-js/src/modules/networking/index.test.js @@ -239,27 +239,78 @@ describe('getEpoxyClient', () => { expect(mockInitEpoxy).toHaveBeenCalledTimes(2); }); - // A failed init must not be cached, otherwise every later socket would keep - // resolving the same broken attempt. + // Swallowing the reason here used to hand callers an undefined client, so + // a dead relay surfaced as "cannot read properties of undefined" instead. + it('surfaces why the client could not be built', async () => { + mockInitEpoxy.mockRejectedValue(new Error('wasm unavailable')); + + await expect(getEpoxyClient()).rejects.toThrow('wasm unavailable'); + }); + + it('surfaces a credential failure the same way', async () => { + mockFetch.mockImplementation(async () => relayResponse({ status: 500 })); + + await expect(getEpoxyClient()).rejects.toThrow(/HTTP 500/); + }); + + it('rejects every caller waiting on a failed attempt', async () => { + mockInitEpoxy.mockRejectedValue(new Error('wasm unavailable')); + + const results = await Promise.allSettled([getEpoxyClient(), getEpoxyClient()]); + + expect(results.map(r => r.status)).toEqual(['rejected', 'rejected']); + expect(mockInitEpoxy).toHaveBeenCalledTimes(1); + }); + + // A failed attempt must not be cached, otherwise every later socket would + // keep resolving the same broken client. it('does not cache a failed init, so the next caller retries', async () => { mockInitEpoxy.mockRejectedValueOnce(new Error('wasm unavailable')); - await getEpoxyClient(); + await expect(getEpoxyClient()).rejects.toThrow('wasm unavailable'); const retried = await getEpoxyClient(); expect(mockInitEpoxy).toHaveBeenCalledTimes(2); expect(retried).toEqual({ client: 'epoxy' }); }); - // Credential failures are swallowed the same way an init failure is. it('does not cache a failed credential fetch', async () => { mockFetch.mockImplementationOnce(async () => relayResponse({ status: 500 })); - await getEpoxyClient(); + await expect(getEpoxyClient()).rejects.toThrow(/HTTP 500/); const retried = await 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); + + const first = getEpoxyClient(); + globalThis.puter.authToken = 'different-tok'; + const second = getEpoxyClient(); + gate.resolve({ client: 'first' }); + + expect(await first).toEqual({ client: 'first' }); + expect(await second).toEqual({ client: 'epoxy' }); + expect(mockInitEpoxy).toHaveBeenCalledTimes(2); + }); }); describe('netAPI surface', () => { @@ -270,3 +321,43 @@ describe('netAPI surface', () => { expect(typeof netAPI.generateWispV1URL).toBe('function'); }); }); + +// The PSocket unit tests stub this module out, so nothing there would notice if +// the two drifted apart. These drive the real socket against the real client +// cache -- only the wasm bundle and the token endpoint are stubbed -- and pin +// the property that matters: a caller learns why the relay is unreachable. +describe('failure reporting through a real socket', () => { + const listen = () => { + const socket = new netAPI.Socket('example.com', 80); + const events = { error: vi.fn(), close: vi.fn() }; + socket.on('error', events.error); + socket.on('close', events.close); + return events; + }; + + const closed = events => + vi.waitFor(() => expect(events.close).toHaveBeenCalled(), { interval: 1 }); + + it('reports why the epoxy client could not be built', async () => { + mockInitEpoxy.mockRejectedValue(new Error('wasm unavailable')); + + const events = listen(); + await closed(events); + + expect(events.error).toHaveBeenCalledTimes(1); + const [reason] = events.error.mock.calls[0]; + expect(reason).toBeInstanceOf(Error); + expect(reason.message).toBe('wasm unavailable'); + expect(events.close).toHaveBeenCalledWith(true); + }); + + it('reports why the relay refused to mint a token', async () => { + mockFetch.mockImplementation(async () => relayResponse({ status: 503 })); + + const events = listen(); + await closed(events); + + expect(events.error.mock.calls[0][0].message).toMatch(/HTTP 503/); + expect(events.close).toHaveBeenCalledWith(true); + }); +});