chadtmiller.com/pds.js

Switch repository

Loading…

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