// Firehose subscribers are held as hibernated WebSockets, which outlive a // restart of the Durable Object. A subscriber whose end has gone must be let // go of: an unclosed socket stays in getWebSockets() and keeps absorbing // broadcasts, so the server holds a connection it can no longer deliver on // while the subscriber, never told to go away, never reconnects — federation // stops with both ends believing they are healthy. import { describe, expect, it } from 'vitest'; import { createWebSocket, PDSDurableObject } from '../src/index.js'; /** * A socket that records how it was closed, and can fail its sends. * @param {{sendFails?: boolean}} [opts] */ function fakeSocket({ sendFails = false } = {}) { return { sent: /** @type {any[]} */ ([]), closed: /** @type {{code: number, reason: string}|null} */ (null), /** @param {any} data */ send(data) { if (sendFails) throw new Error('socket is gone'); this.sent.push(data); }, /** @param {number} code @param {string} reason */ close(code, reason) { this.closed = { code, reason }; }, }; } /** * The real WebSocket port, over a fixed set of sockets. * @param {any[]} sockets */ function port(sockets) { return /** @type {any} */ ( createWebSocket(/** @type {any} */ ({ getWebSockets: () => sockets })) ); } // Firehose frames are bytes, not text. const EVENT = new Uint8Array([0xa1, 0x62, 0x6f, 0x70]); describe('broadcasting to subscribers', () => { it('drops a subscriber whose send fails, so it is asked to reconnect', () => { const live = fakeSocket(); const gone = fakeSocket({ sendFails: true }); port([live, gone]).broadcast(EVENT); expect(live.sent).toEqual([EVENT]); expect(live.closed).toBeNull(); // The dead one is closed rather than silently retained. expect(gone.closed?.code).toBe(1011); }); it('keeps delivering to healthy subscribers past a dead one', () => { const gone = fakeSocket({ sendFails: true }); const live = fakeSocket(); port([gone, live]).broadcast(EVENT); expect(live.sent).toEqual([EVENT]); }); it('survives a socket that fails both send and close', () => { const hostile = { send() { throw new Error('gone'); }, close() { throw new Error('also gone'); }, }; const live = fakeSocket(); expect(() => port([hostile, live]).broadcast(EVENT)).not.toThrow(); expect(live.sent).toEqual([EVENT]); }); }); describe('WebSocket lifecycle callbacks', () => { it('completes the closing handshake so the socket leaves getWebSockets()', async () => { const ws = fakeSocket(); await /** @type {any} */ (PDSDurableObject.prototype).webSocketClose.call( {}, ws, 1001, 'going away', true, ); expect(ws.closed).toEqual({ code: 1001, reason: 'going away' }); }); it('releases a socket that errored', async () => { const ws = fakeSocket(); await /** @type {any} */ (PDSDurableObject.prototype).webSocketError.call( {}, ws, new Error('boom'), ); expect(ws.closed?.code).toBe(1011); }); it('does not throw when the socket is already closed', async () => { const ws = { send() {}, close() { throw new Error('already closed'); }, }; await expect( /** @type {any} */ (PDSDurableObject.prototype).webSocketClose.call( {}, ws, 1000, '', true, ), ).resolves.toBeUndefined(); await expect( /** @type {any} */ (PDSDurableObject.prototype).webSocketError.call( {}, ws, new Error('boom'), ), ).resolves.toBeUndefined(); }); });