fix(networking): recover shared relay clients and preserve write order (#4377)

* fix(networking): recover shared relay clients and preserve write order

* fix(networking): clear the flush flag with the last queue check

A write landing between the drained queue and the connect path resuming
was queued with nothing left to send it.

* fix(networking): keep a throwing socket listener out of connect and close

An open listener that threw was caught as a failed connect and could
retry onto a second stream; an error listener that threw stopped the
socket from ever closing. Both now rethrow on their own tick.

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;

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();

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)],

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);
+    });
+});

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<object, { transport: import('./types.js').RelayTransport }>} */
+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.
  *

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 }));
 

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,

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(),
+        );
     });
 });

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<WebSocket>;
+ *     current: WebSocket | undefined;
+ *     down: boolean;
+ *     closed: boolean;
+ * }} RelayTransport
+ * @internal
+ */
+
 export {};