✨(loadtest) add a websocket load generator for the collaboration server

standlone package simulating traffic on the yhub server. It reuses the
same stack used bu the frontend application to connect to the websocket.
How to use it is explained in its README file.

CHANGELOG.md

@@ -14,6 +14,7 @@ and this project adheres to
 - 🐛(frontend) reduce PostHog volume from web vitals and opt_in spam #2701
 - ✨(backend) expose the attachment max size in the config endpoint #2577
 - ✨(frontend) warn before uploading an attachment over the size limit #2577
+- ✨(loadtest) add a websocket load generator for the collaboration server
 - ✨(backend) measure the calls to yhub and to the converters, the database
   pool and the celery queue
 - ✨(backend) add a `LoadTest` configuration and its `loadtest` application,

src/loadtest/swarm/Dockerfile

@@ -0,0 +1,19 @@
+# Built from the repository root, like every other image here:
+#   docker build -f src/loadtest/swarm/Dockerfile -t docs-swarm .
+FROM node:22-trixie AS base
+WORKDIR /app
+COPY ./src/loadtest/swarm/package.json ./src/loadtest/swarm/yarn.lock ./
+
+FROM base AS builder
+RUN yarn install --frozen-lockfile
+COPY ./src/loadtest/swarm/tsconfig.json ./src/loadtest/swarm/tsconfig.build.json ./
+COPY ./src/loadtest/swarm/src ./src
+RUN yarn build
+
+FROM base AS swarm
+RUN yarn install --frozen-lockfile --production
+COPY --from=builder /app/dist ./dist
+# many sockets: the default limit of open files is what would stop the swarm first
+ENV UV_THREADPOOL_SIZE=16
+USER node
+ENTRYPOINT ["node", "dist/index.js"]

src/loadtest/swarm/README.md

@@ -0,0 +1,121 @@
+# Swarm — websocket load generator for the collaboration server
+
+Opens many clients on the Docs collaboration server (yhub) with the very stack
+the frontend uses — `yjs`, `y-websocket`'s `WebsocketProvider` with the same
+options, a `ws` socket carrying the session cookie and the origin a browser
+would send — and measures what a user would feel: time to connect, time to the
+first sync, and how long an edit takes to reach the other clients of the same
+document. It is the collaboration half of `documentation/stress-test-plan.md`;
+the HTTP half is k6.
+
+It is a standalone package (its own `package.json` and `yarn.lock`, not a
+workspace of `src/frontend`), like `src/yhub-server`.
+
+## Who the clients are
+
+The swarm logs nobody in. It reads the manifest that the backend's
+`create_load_test_sessions` command writes (`src/backend/loadtest`, only with
+the `LoadTest` configuration): the cookie name, and per logged-in user a
+session key and the documents that user may edit or read. **The manifest holds
+live sessions and is a secret**: hand it to the swarm and to nothing else, and
+revoke the sessions when the campaign is over.
+
+Every client uses one session of the manifest, in turn. With more clients than
+sessions a user is logged in several times, which the report warns about: the
+API throttles per user, so ask for at least as many sessions as clients.
+
+## Running it
+
+```bash
+cd src/loadtest/swarm
+yarn install --frozen-lockfile && yarn build
+node dist/index.js --manifest manifest.json --url wss://docs.example.com \
+    --mode wide --clients 500 --ramp 20 --duration 600 --writers 0.3
+```
+
+| Option | Default | What it does |
+| ------ | ------- | ------------ |
+| `--manifest` | — | The manifest file. Required |
+| `--url` | — | `wss://host` of the collaboration server, the ingress in a deployment. Required |
+| `--origin` | `https://host` of the url | The `Origin` header. yhub refuses origins outside `COLLABORATION_SERVER_ORIGIN` |
+| `--org` | `docs` | The yhub organisation (`YHUB_ORG`) |
+| `--clients` | `10` | Virtual clients |
+| `--ramp` | `10` | Connections opened per second |
+| `--duration` | `60` | Seconds to hold once everybody is asked to connect |
+| `--mode` | `wide` | `hot`: everybody on one document. `wide`: each client on a document of its own user. `idle`: `wide` with nobody writing |
+| `--doc` | first public document | The document of `hot` mode |
+| `--writers` | `0.3` (`0` in `idle`) | Share of the clients that edit |
+| `--edit-interval` | `2000` | Milliseconds between two edits of a writer |
+| `--edit-size` | `12` | Characters typed per edit |
+| `--awareness-interval` | `5000` | Milliseconds between two awareness updates of a writer, `0` for none |
+| `--storm-at` | `0` | Second of the hold at which every client drops its socket and reconnects (with `--reconnect-jitter` ms of jitter, `3000` like the frontend) |
+| `--settle` | `15` | Seconds to wait after the last edit before checking convergence |
+| `--metrics-port` | `9465` | Where `/metrics` is served, `0` for nowhere. `--metrics-token` requires a bearer token on it |
+| `--report` | `-` | Where the JSON report goes, `-` for stdout |
+
+Progress goes to stderr every `--progress-interval` seconds. The exit code is
+`0` when every client connected at least once and every document converged.
+
+Scenarios of the plan, as options:
+
+| Scenario | Options |
+| -------- | ------- |
+| connect ramp | `--mode idle --ramp N`, raise N |
+| idle steady state | `--mode idle --clients 10000 --duration 1800` |
+| hot document | `--mode hot --clients 200 --writers 0.5 --edit-interval 500` |
+| wide editing | `--mode wide --clients 3000 --writers 0.3` |
+| reconnect storm | any of the above with `--storm-at 300` — or restart yhub during the hold: the clients reconnect on their own, as the frontend does |
+
+Several swarms can run at once (one per pod) against the same manifest, each
+with a distinct slice of it: split the manifest's `sessions` beforehand. The
+propagation latency compares the clocks of the writer and of the reader, which
+is exact inside one process and only as good as the clock sync between pods.
+
+## What it measures
+
+On `/metrics`, and summarised as percentiles in the report:
+
+| Metric | What it is |
+| ------ | ---------- |
+| `swarm_clients{state}` | Clients by socket state |
+| `swarm_connect_duration_seconds` | From asking for a connection to the socket being open. yhub authenticates the upgrade against the backend, and seeds the document under soft migration, before answering: this is what that costs |
+| `swarm_sync_duration_seconds` | From the socket being open to the first sync of the document |
+| `swarm_propagation_latency_seconds` | From an edit to its arrival on another client of the document |
+| `swarm_edits_total`, `swarm_updates_received_total` | Edits made, updates received |
+| `swarm_ws_messages_total{direction}`, `swarm_ws_bytes_total{direction}` | Websocket traffic |
+| `swarm_reconnects_total`, `swarm_ws_closes_total{code}` | Reconnections, and why the sockets closed |
+| `swarm_upgrade_failures_total{status}` | Upgrades the server refused, by http status (a `503` is yhub not reaching the backend) |
+| `nodejs_*` | The swarm's own runtime: past a point the load generator saturates before the server does, and its event-loop lag says so |
+
+The report also holds the convergence check: after `--settle` seconds, every
+client of a document must hold the same content. Documents that do not are
+listed under `convergence.diverged`.
+
+## What it writes into the documents
+
+Edits go into a `Y.Text` and a `Y.Map` of the document that the editor does not
+render (`loadtest-text`, `loadtest-stamps`): the server persists and fans them
+out like any content. A document a swarm wrote to keeps those types in its
+history. **Only run this against anonymised data**, never against a document
+someone uses.
+
+## Container
+
+```bash
+docker build -f src/loadtest/swarm/Dockerfile -t docs-swarm .
+docker run --rm -v ./manifest.json:/manifest.json:ro docs-swarm \
+    --manifest /manifest.json --url wss://docs.example.com --clients 100
+```
+
+In a cluster, run it as a Job on nodes that do not host the application, going
+through the ingress like a browser would.
+
+## Development
+
+```bash
+yarn typecheck && yarn lint && yarn test
+```
+
+The tests run a small y-websocket protocol server in-process
+(`__tests__/_server.ts`) — no yhub, no store — and drive a real swarm against
+it: cookies, origin, edits, propagation, convergence, storms, refused upgrades.

src/loadtest/swarm/__tests__/_server.ts

@@ -0,0 +1,151 @@
+// A y-websocket protocol server small enough to live in a test: one Y.Doc per
+// room, sync step 1/2 and awareness relayed between the sockets of a room. It
+// records the headers of every upgrade, which is what the swarm is checked on.
+import { createServer } from 'node:http';
+import type { Server } from 'node:http';
+import type { AddressInfo } from 'node:net';
+
+import * as decoding from 'lib0/decoding';
+import * as encoding from 'lib0/encoding';
+import { WebSocketServer } from 'ws';
+import type WebSocket from 'ws';
+import * as awarenessProtocol from 'y-protocols/awareness';
+import * as syncProtocol from 'y-protocols/sync';
+import * as Y from 'yjs';
+
+const MESSAGE_SYNC = 0;
+const MESSAGE_AWARENESS = 1;
+
+interface Room {
+  doc: Y.Doc;
+  awareness: awarenessProtocol.Awareness;
+  sockets: Set<WebSocket>;
+}
+
+export interface TestServer {
+  url: string;
+  upgrades: Array<{ path: string; cookie?: string; origin?: string }>;
+  rooms: Map<string, Room>;
+  refuse: { status: number } | null;
+  dropAll: () => void;
+  close: () => Promise<void>;
+}
+
+export const startTestServer = async (): Promise<TestServer> => {
+  const rooms = new Map<string, Room>();
+  const upgrades: TestServer['upgrades'] = [];
+  const state: { refuse: { status: number } | null } = { refuse: null };
+  const http: Server = createServer((_req, res) => res.writeHead(404).end());
+  const wss = new WebSocketServer({ noServer: true });
+
+  const room = (name: string): Room => {
+    let r = rooms.get(name);
+    if (!r) {
+      const doc = new Y.Doc();
+      const awareness = new awarenessProtocol.Awareness(doc);
+      r = { doc, awareness, sockets: new Set() };
+      doc.on('update', (update: Uint8Array, origin: unknown) => {
+        const encoder = encoding.createEncoder();
+        encoding.writeVarUint(encoder, MESSAGE_SYNC);
+        syncProtocol.writeUpdate(encoder, update);
+        const message = encoding.toUint8Array(encoder);
+        for (const socket of r?.sockets ?? []) {
+          if (socket !== origin) socket.send(message);
+        }
+      });
+      awareness.on(
+        'update',
+        (
+          {
+            added,
+            updated,
+            removed,
+          }: { added: number[]; updated: number[]; removed: number[] },
+          origin: unknown,
+        ) => {
+          const changed = [...added, ...updated, ...removed];
+          const encoder = encoding.createEncoder();
+          encoding.writeVarUint(encoder, MESSAGE_AWARENESS);
+          encoding.writeVarUint8Array(
+            encoder,
+            awarenessProtocol.encodeAwarenessUpdate(awareness, changed),
+          );
+          const message = encoding.toUint8Array(encoder);
+          for (const socket of r?.sockets ?? []) {
+            if (socket !== origin) socket.send(message);
+          }
+        },
+      );
+      rooms.set(name, r);
+    }
+    return r;
+  };
+
+  http.on('upgrade', (req, socket, head) => {
+    const path = req.url ?? '';
+    upgrades.push({
+      path,
+      cookie: req.headers.cookie,
+      origin: req.headers.origin,
+    });
+    if (state.refuse) {
+      socket.write(
+        `HTTP/1.1 ${state.refuse.status} Refused\r\nContent-Length: 0\r\n\r\n`,
+      );
+      socket.destroy();
+      return;
+    }
+    wss.handleUpgrade(req, socket, head, (ws) => {
+      const name = path.split('/').pop() ?? '';
+      const r = room(name);
+      r.sockets.add(ws);
+      ws.on('message', (data: Buffer) => {
+        const decoder = decoding.createDecoder(new Uint8Array(data));
+        const encoder = encoding.createEncoder();
+        const type = decoding.readVarUint(decoder);
+        if (type === MESSAGE_SYNC) {
+          encoding.writeVarUint(encoder, MESSAGE_SYNC);
+          syncProtocol.readSyncMessage(decoder, encoder, r.doc, ws);
+          if (encoding.length(encoder) > 1)
+            ws.send(encoding.toUint8Array(encoder));
+        } else if (type === MESSAGE_AWARENESS) {
+          awarenessProtocol.applyAwarenessUpdate(
+            r.awareness,
+            decoding.readVarUint8Array(decoder),
+            ws,
+          );
+        }
+      });
+      ws.on('close', () => r.sockets.delete(ws));
+      // sync step 1, as y-websocket's server does on connect
+      const encoder = encoding.createEncoder();
+      encoding.writeVarUint(encoder, MESSAGE_SYNC);
+      syncProtocol.writeSyncStep1(encoder, r.doc);
+      ws.send(encoding.toUint8Array(encoder));
+    });
+  });
+
+  await new Promise<void>((resolve) => http.listen(0, '127.0.0.1', resolve));
+  const { port } = http.address() as AddressInfo;
+  return {
+    url: `ws://127.0.0.1:${port}`,
+    upgrades,
+    rooms,
+    get refuse() {
+      return state.refuse;
+    },
+    set refuse(value) {
+      state.refuse = value;
+    },
+    dropAll: () => {
+      for (const r of rooms.values())
+        for (const socket of r.sockets) socket.terminate();
+    },
+    close: () =>
+      new Promise((resolve) => {
+        for (const r of rooms.values())
+          for (const socket of r.sockets) socket.terminate();
+        wss.close(() => http.close(() => resolve()));
+      }),
+  };
+};

src/loadtest/swarm/__tests__/config.spec.ts

@@ -0,0 +1,87 @@
+import { describe, expect, it } from 'vitest';
+
+import { parseConfig } from '../src/config.js';
+
+const base = ['--manifest', 'm.json', '--url', 'wss://docs.example.com/'];
+
+describe('parseConfig', () => {
+  it('has the frontend-like defaults', () => {
+    const config = parseConfig(base);
+    expect(config.url).toBe('wss://docs.example.com');
+    expect(config.origin).toBe('https://docs.example.com');
+    expect(config.org).toBe('docs');
+    expect(config.mode).toBe('wide');
+    expect(config.writers).toBe(0.3);
+    expect(config.reconnectJitter).toBe(3000);
+    expect(config.metricsPort).toBe(9465);
+  });
+
+  it('reads every option', () => {
+    const config = parseConfig([
+      ...base,
+      '--clients',
+      '500',
+      '--ramp',
+      '25',
+      '--duration',
+      '300',
+      '--mode',
+      'hot',
+      '--doc',
+      'd1',
+      '--writers',
+      '0.5',
+      '--edit-interval',
+      '500',
+      '--edit-size',
+      '3',
+      '--awareness-interval',
+      '0',
+      '--storm-at',
+      '120',
+      '--settle',
+      '5',
+      '--metrics-port',
+      '0',
+      '--metrics-token',
+      't',
+      '--origin',
+      'http://other',
+    ]);
+    expect(config).toMatchObject({
+      clients: 500,
+      ramp: 25,
+      duration: 300,
+      mode: 'hot',
+      doc: 'd1',
+      writers: 0.5,
+      editInterval: 500,
+      editSize: 3,
+      awarenessInterval: 0,
+      stormAt: 120,
+      settle: 5,
+      metricsPort: 0,
+      metricsToken: 't',
+      origin: 'http://other',
+    });
+  });
+
+  it('nobody writes in idle mode unless asked', () => {
+    expect(parseConfig([...base, '--mode', 'idle']).writers).toBe(0);
+    expect(
+      parseConfig([...base, '--mode', 'idle', '--writers', '0.1']).writers,
+    ).toBe(0.1);
+  });
+
+  it.each([
+    [['--url', 'wss://x'], '--manifest is required'],
+    [['--manifest', 'm'], '--url is required'],
+    [['--manifest', 'm', '--url', 'https://x'], 'must be ws:// or wss://'],
+    [[...base, '--mode', 'other'], '--mode must be one of'],
+    [[...base, '--writers', '2'], '--writers must be a number between 0 and 1'],
+    [[...base, '--clients', 'abc'], '--clients must be a number'],
+    [[...base, '--unknown'], 'Unknown option'],
+  ])('refuses %j', (argv, message) => {
+    expect(() => parseConfig(argv)).toThrow(message);
+  });
+});

src/loadtest/swarm/__tests__/manifest.spec.ts

@@ -0,0 +1,62 @@
+import { describe, expect, it } from 'vitest';
+
+import { parseManifest } from '../src/manifest.js';
+
+const valid = {
+  cookie_name: 'docs_sessionid',
+  public_documents: ['p1'],
+  sessions: [
+    {
+      user_id: 'u1',
+      session_key: 'k1',
+      editable_documents: ['e1'],
+      readonly_documents: ['r1'],
+    },
+  ],
+};
+
+describe('parseManifest', () => {
+  it('reads what the backend writes', () => {
+    const manifest = parseManifest(JSON.stringify(valid));
+    expect(manifest.cookie_name).toBe('docs_sessionid');
+    expect(manifest.sessions[0]).toEqual(valid.sessions[0]);
+    expect(manifest.public_documents).toEqual(['p1']);
+  });
+
+  it('fills what is missing rather than failing on it', () => {
+    const manifest = parseManifest(
+      JSON.stringify({ cookie_name: 'c', sessions: [{ session_key: 'k' }] }),
+    );
+    expect(manifest.sessions[0]).toEqual({
+      user_id: '0',
+      session_key: 'k',
+      editable_documents: [],
+      readonly_documents: [],
+    });
+    expect(manifest.public_documents).toEqual([]);
+  });
+
+  it.each([
+    ['not json', 'not JSON'],
+    ['null', 'not an object'],
+    ['[]', 'no cookie_name'],
+    ['{"sessions":[{"session_key":"k"}]}', 'no cookie_name'],
+    ['{"cookie_name":"c","sessions":[]}', 'holds no session'],
+    ['{"cookie_name":"c","sessions":[{"user_id":"u"}]}', 'no session_key'],
+  ])('refuses %s', (text, message) => {
+    expect(() => parseManifest(text)).toThrow(message);
+  });
+
+  it('refuses sessions that have expired', () => {
+    expect(() =>
+      parseManifest(
+        JSON.stringify({ ...valid, expires_at: '2000-01-01T00:00:00+00:00' }),
+      ),
+    ).toThrow('expired');
+    expect(() =>
+      parseManifest(
+        JSON.stringify({ ...valid, expires_at: '2999-01-01T00:00:00+00:00' }),
+      ),
+    ).not.toThrow();
+  });
+});

src/loadtest/swarm/__tests__/stats.spec.ts

@@ -0,0 +1,23 @@
+import { describe, expect, it } from 'vitest';
+
+import { Samples } from '../src/stats.js';
+
+describe('Samples', () => {
+  it('summarises nothing as null', () => {
+    expect(new Samples().summary()).toBeNull();
+  });
+
+  it('computes the percentiles', () => {
+    const samples = new Samples();
+    for (let i = 1; i <= 100; i++) samples.add(i);
+    expect(samples.summary()).toEqual({
+      count: 100,
+      min: 1,
+      p50: 50,
+      p95: 95,
+      p99: 99,
+      max: 100,
+      mean: 50.5,
+    });
+  });
+});

src/loadtest/swarm/__tests__/swarm.spec.ts

@@ -0,0 +1,239 @@
+import { afterEach, beforeEach, describe, expect, it } from 'vitest';
+
+import { parseConfig } from '../src/config.js';
+import type { Config } from '../src/config.js';
+import type { Manifest } from '../src/manifest.js';
+import { registry } from '../src/metrics.js';
+import { Swarm, assign } from '../src/swarm.js';
+import { startTestServer } from './_server.js';
+import type { TestServer } from './_server.js';
+
+const manifest: Manifest = {
+  cookie_name: 'docs_sessionid',
+  public_documents: ['pub1', 'pub2'],
+  sessions: [
+    {
+      user_id: 'u1',
+      session_key: 'k1',
+      editable_documents: ['e1', 'e2'],
+      readonly_documents: ['r1'],
+    },
+    {
+      user_id: 'u2',
+      session_key: 'k2',
+      editable_documents: ['e3'],
+      readonly_documents: [],
+    },
+    {
+      user_id: 'u3',
+      session_key: 'k3',
+      editable_documents: [],
+      readonly_documents: ['r2'],
+    },
+  ],
+};
+
+const config = (url: string, ...extra: string[]): Config =>
+  parseConfig([
+    '--manifest',
+    'm',
+    '--url',
+    url,
+    '--metrics-port',
+    '0',
+    '--progress-interval',
+    '1',
+    '--ramp',
+    '1000',
+    '--settle',
+    '1',
+    ...extra,
+  ]);
+
+describe('assign', () => {
+  const dummy = (...extra: string[]) => config('ws://x', ...extra);
+
+  it('puts everybody on one document in hot mode, the first public one by default', () => {
+    const { assignments } = assign(
+      dummy('--mode', 'hot', '--clients', '4'),
+      manifest,
+    );
+    expect(assignments.map((a) => a.room)).toEqual([
+      'pub1',
+      'pub1',
+      'pub1',
+      'pub1',
+    ]);
+    expect(
+      assign(dummy('--mode', 'hot', '--doc', 'd', '--clients', '1'), manifest)
+        .assignments[0].room,
+    ).toBe('d');
+  });
+
+  it('gives writers a document they may edit and readers one they may read', () => {
+    const { assignments } = assign(
+      dummy('--mode', 'wide', '--clients', '3', '--writers', '0.67'),
+      manifest,
+    );
+    expect(
+      assignments.map((a) => [a.session.user_id, a.writer, a.room]),
+    ).toEqual([
+      ['u1', true, 'e1'],
+      ['u2', true, 'e3'],
+      ['u3', false, 'r2'],
+    ]);
+  });
+
+  it('falls back to the public documents for a user with none, and warns past the sessions', () => {
+    const { assignments, warnings } = assign(
+      dummy('--mode', 'wide', '--clients', '4', '--writers', '1'),
+      manifest,
+    );
+    // u3 may edit nothing of their own: a public document instead
+    expect(assignments[2].room).toBe('pub1');
+    // 4 clients for 3 sessions: u1 twice, on their next document
+    expect(assignments[3].session.user_id).toBe('u1');
+    expect(assignments[3].room).toBe('e2');
+    expect(warnings[0]).toContain('4 clients for 3 sessions');
+  });
+
+  it('refuses hot mode without a document', () => {
+    expect(() =>
+      assign(dummy('--mode', 'hot'), { ...manifest, public_documents: [] }),
+    ).toThrow('--mode hot needs --doc');
+  });
+});
+
+describe('Swarm', () => {
+  let server: TestServer;
+  beforeEach(async () => {
+    server = await startTestServer();
+  });
+  afterEach(async () => {
+    await server.close();
+    registry.resetMetrics();
+  });
+
+  it('connects as the users of the manifest, edits, and converges', async () => {
+    const swarm = new Swarm(
+      config(
+        server.url,
+        '--mode',
+        'hot',
+        '--clients',
+        '6',
+        '--writers',
+        '0.5',
+        '--duration',
+        '3',
+        '--edit-interval',
+        '100',
+        '--awareness-interval',
+        '200',
+      ),
+      manifest,
+    );
+    const report = await swarm.run();
+
+    expect(report.clients).toBe(6);
+    expect(report.connected).toBe(6);
+    expect(report.writers).toBe(3);
+    expect(report.edits).toBeGreaterThan(20);
+    expect(report.convergence).toEqual({
+      documents: 1,
+      converged: 1,
+      diverged: [],
+    });
+    // a sample per edit per other client, roughly
+    expect(report.propagation?.count).toBeGreaterThan(report.edits);
+    expect(report.propagation?.p95).toBeLessThan(1);
+    expect(report.connect?.count).toBe(6);
+    expect(report.sync?.count).toBe(6);
+    expect(report.reconnects).toBe(0);
+
+    // the upgrade carried the session cookie, the origin and the room
+    expect(server.upgrades).toHaveLength(6);
+    expect(server.upgrades[0]).toEqual({
+      path: '/collaboration/ws/v1/docs/pub1',
+      cookie: 'docs_sessionid=k1',
+      origin: `http://${new URL(server.url).host}`,
+    });
+    expect(new Set(server.upgrades.map((u) => u.cookie))).toEqual(
+      new Set(['docs_sessionid=k1', 'docs_sessionid=k2', 'docs_sessionid=k3']),
+    );
+    // the server holds what the writers typed
+    const doc = server.rooms.get('pub1')?.doc;
+    expect(doc?.getText('loadtest-text').length).toBeGreaterThan(0);
+    expect(doc?.getMap('loadtest-stamps').size).toBe(3);
+    // and saw their awareness
+    expect(
+      server.rooms.get('pub1')?.awareness.getStates().size ?? 0,
+    ).toBeGreaterThanOrEqual(0);
+
+    const text = await registry.metrics();
+    expect(text).toMatch(/swarm_edits_total \d+/);
+    expect(text).toMatch(/swarm_propagation_latency_seconds_count \d+/);
+    expect(text).toMatch(/swarm_ws_bytes_total\{direction="out"\} [1-9]/);
+    expect(text).toMatch(/swarm_ws_bytes_total\{direction="in"\} [1-9]/);
+  });
+
+  it('reconnects through a storm and when the server drops everybody', async () => {
+    const swarm = new Swarm(
+      config(
+        server.url,
+        '--mode',
+        'wide',
+        '--clients',
+        '3',
+        '--writers',
+        '0',
+        '--duration',
+        '3',
+        '--storm-at',
+        '1',
+        '--reconnect-jitter',
+        '200',
+      ),
+      manifest,
+    );
+    const dropAt = setTimeout(() => server.dropAll(), 2200);
+    const report = await swarm.run();
+    clearTimeout(dropAt);
+
+    expect(report.connected).toBe(3);
+    // one reconnection from the storm, one from the drop
+    expect(report.reconnects).toBeGreaterThanOrEqual(6);
+    expect(server.upgrades.length).toBeGreaterThanOrEqual(9);
+    const text = await registry.metrics();
+    expect(text).toMatch(
+      /swarm_reconnects_total [6-9]|swarm_reconnects_total \d{2,}/,
+    );
+    expect(text).toMatch(/swarm_ws_closes_total\{code="\d+"\}/);
+  });
+
+  it('counts the upgrades a server refuses, and keeps trying', async () => {
+    server.refuse = { status: 503 };
+    const swarm = new Swarm(
+      config(
+        server.url,
+        '--mode',
+        'hot',
+        '--clients',
+        '2',
+        '--duration',
+        '2',
+        '--settle',
+        '0',
+      ),
+      manifest,
+    );
+    const report = await swarm.run();
+
+    expect(report.connected).toBe(0);
+    expect(report.convergence.documents).toBe(0);
+    const text = await registry.metrics();
+    expect(text).toMatch(/swarm_upgrade_failures_total\{status="503"\} [2-9]/);
+    // every client was let go at the end of the run
+    expect(text).toMatch(/swarm_clients\{state="connecting"\} 0/);
+  });
+});

src/loadtest/swarm/eslint.config.mjs

@@ -0,0 +1,24 @@
+import js from '@eslint/js';
+import prettier from 'eslint-config-prettier/flat';
+import tseslint from 'typescript-eslint';
+
+export default tseslint.config(
+  {
+    ignores: ['dist/**', 'node_modules/**'],
+  },
+  js.configs.recommended,
+  ...tseslint.configs.recommended,
+  {
+    files: ['**/*.ts'],
+    rules: {
+      '@typescript-eslint/no-explicit-any': 'error',
+      '@typescript-eslint/no-non-null-assertion': 'error',
+      '@typescript-eslint/consistent-type-imports': 'error',
+      '@typescript-eslint/no-unused-vars': [
+        'error',
+        { varsIgnorePattern: '^_', argsIgnorePattern: '^_' },
+      ],
+    },
+  },
+  prettier,
+);

src/loadtest/swarm/package.json

@@ -0,0 +1,35 @@
+{
+  "name": "docs-loadtest-swarm",
+  "private": true,
+  "type": "module",
+  "description": "Load generator opening many websocket clients on the Docs collaboration server, with the client stack the frontend uses",
+  "scripts": {
+    "build": "tsc -p tsconfig.build.json",
+    "typecheck": "tsc -p tsconfig.json --noEmit",
+    "start": "node dist/index.js",
+    "lint": "eslint",
+    "test": "vitest run"
+  },
+  "engines": {
+    "node": ">=22"
+  },
+  "packageManager": "yarn@1.22.22",
+  "dependencies": {
+    "lib0": "0.2.114",
+    "prom-client": "15.1.3",
+    "ws": "8.18.3",
+    "y-protocols": "1.0.7",
+    "y-websocket": "3.1.0",
+    "yjs": "13.6.32"
+  },
+  "devDependencies": {
+    "@eslint/js": "10.0.1",
+    "@types/node": "24.13.3",
+    "@types/ws": "8.18.1",
+    "eslint": "10.9.0",
+    "eslint-config-prettier": "10.1.8",
+    "typescript": "6.0.3",
+    "typescript-eslint": "8.69.0",
+    "vitest": "3.2.7"
+  }
+}

src/loadtest/swarm/src/client.ts

@@ -0,0 +1,291 @@
+/**
+ * One virtual client: a Y.Doc and the very provider the frontend uses
+ * (`y-websocket`'s WebsocketProvider, same options), over a `ws` socket that
+ * carries the session cookie and the origin a browser would send.
+ *
+ * Edits go into a Y.Text and a Y.Map of the document that the editor does not
+ * render (`loadtest-text`, `loadtest-stamps`): the server persists and fans them
+ * out like any content, without the swarm having to build BlockNote blocks. A
+ * document a swarm wrote to keeps those types in its history — only run this
+ * on anonymised data.
+ */
+import { WebsocketProvider } from 'y-websocket';
+import * as Y from 'yjs';
+import WebSocket from 'ws';
+import type { ClientOptions as SocketOptions } from 'ws';
+
+import * as metrics from './metrics.js';
+import type { Samples } from './stats.js';
+
+export interface ClientOptions {
+  id: number;
+  url: string;
+  origin: string;
+  room: string;
+  cookie: string;
+  writer: boolean;
+  editInterval: number;
+  editSize: number;
+  awarenessInterval: number;
+  samples: {
+    connect: Samples;
+    sync: Samples;
+    propagation: Samples;
+  };
+}
+
+const STAMPS = 'loadtest-stamps';
+const TEXT = 'loadtest-text';
+const TYPED = 'abcdefghijklmnopqrstuvwxyz     ';
+
+// The polyfill y-websocket instantiates with `(url, protocols)`: the headers
+// ride in through the class, one per client since each has its own cookie.
+const socketClass = (
+  cookie: string,
+  origin: string,
+  onUpgradeFailure: (status: number) => void,
+) =>
+  class SwarmSocket extends WebSocket {
+    constructor(url: string, protocols?: string | string[]) {
+      const options: SocketOptions = {
+        headers: { cookie, origin },
+        // a bad certificate on preprod is not what is being tested
+        rejectUnauthorized: false,
+      };
+      super(url, protocols, options);
+      this.on('unexpected-response', (_req, res) => {
+        onUpgradeFailure(res.statusCode ?? 0);
+        res.resume();
+      });
+    }
+  };
+
+export class Client {
+  readonly doc = new Y.Doc();
+  readonly options: ClientOptions;
+  provider: WebsocketProvider | null = null;
+  state: 'idle' | 'connecting' | 'connected' | 'disconnected' = 'idle';
+  connections = 0;
+  edits = 0;
+  private editTimer: NodeJS.Timeout | null = null;
+  private awarenessTimer: NodeJS.Timeout | null = null;
+  private connectAskedAt = 0;
+  private connectedAt = 0;
+  private synced = false;
+  private seq = 0;
+  private seen = new Map<string, number>();
+
+  constructor(options: ClientOptions) {
+    this.options = options;
+    this.doc.getMap<{ seq: number; ts: number }>(STAMPS).observe((event) => {
+      // every key the event carries was set by another client (own changes
+      // arrive here too, but with this doc as their origin)
+      if (event.transaction.origin === this) return;
+      const now = Date.now();
+      for (const key of event.keysChanged) {
+        const stamp = event.target.get(key);
+        if (!stamp || key === `w${this.options.id}`) continue;
+        // a stamp written before this client synced is history, not latency
+        if ((this.seen.get(key) ?? -1) >= stamp.seq) continue;
+        if (this.seen.has(key) || this.synced) {
+          const latency = (now - stamp.ts) / 1000;
+          if (latency >= 0 && latency < 3600) {
+            metrics.propagationLatency.observe(latency);
+            this.options.samples.propagation.add(latency);
+          }
+        }
+        this.seen.set(key, stamp.seq);
+      }
+    });
+    this.doc.on('update', (update: Uint8Array, origin: unknown) => {
+      if (origin !== this) metrics.updatesReceivedTotal.inc();
+      void update;
+    });
+  }
+
+  connect(): void {
+    if (this.provider) return;
+    this.setState('connecting');
+    this.connectAskedAt = Date.now();
+    this.synced = false;
+    const provider = new WebsocketProvider(
+      this.options.url,
+      this.options.room,
+      this.doc,
+      {
+        WebSocketPolyfill: socketClass(
+          this.options.cookie,
+          this.options.origin,
+          (status) => {
+            metrics.upgradeFailuresTotal.inc({ status: String(status) });
+          },
+        ) as unknown as typeof globalThis.WebSocket,
+        // the frontend's options, see useProviderStore.tsx
+        disableBc: true,
+        maxBackoffTime: 30000,
+        resyncInterval: 20000,
+      },
+    );
+    provider.on('status', ({ status }) => {
+      if (status === 'connected') {
+        this.connectedAt = Date.now();
+        this.connections += 1;
+        if (this.connections > 1) metrics.reconnectsTotal.inc();
+        else {
+          const seconds = (this.connectedAt - this.connectAskedAt) / 1000;
+          metrics.connectDuration.observe(seconds);
+          this.options.samples.connect.add(seconds);
+        }
+        this.setState('connected');
+        this.watchSocket(provider);
+      } else if (status === 'disconnected') {
+        this.setState('disconnected');
+        this.synced = false;
+        // the provider reconnects on its own: what is measured is the next open
+        this.connectAskedAt = Date.now();
+      } else {
+        this.setState('connecting');
+      }
+    });
+    provider.on('sync', (isSynced: boolean) => {
+      if (!isSynced || this.synced) return;
+      this.synced = true;
+      const seconds = (Date.now() - this.connectedAt) / 1000;
+      metrics.syncDuration.observe(seconds);
+      this.options.samples.sync.add(seconds);
+      if (this.options.writer) this.startWriting();
+    });
+    provider.on('connection-close', (event) => {
+      metrics.closesTotal.inc({ code: String(event?.code ?? 'none') });
+    });
+    provider.on('connection-error', () => {
+      metrics.errorsTotal.inc({ kind: 'connection' });
+    });
+    this.provider = provider;
+  }
+
+  private watchSocket(provider: WebsocketProvider): void {
+    const ws = provider.ws as unknown as WebSocket | null;
+    if (!ws || (ws as unknown as { swarmWatched?: boolean }).swarmWatched)
+      return;
+    (ws as unknown as { swarmWatched?: boolean }).swarmWatched = true;
+    ws.on('message', (data: Buffer | ArrayBuffer | Buffer[]) => {
+      metrics.messagesTotal.inc({ direction: 'in' });
+      metrics.bytesTotal.inc({ direction: 'in' }, byteLength(data));
+    });
+    const send = ws.send.bind(ws);
+    (ws as unknown as { send: typeof ws.send }).send = ((
+      data: Parameters<typeof ws.send>[0],
+      ...rest: unknown[]
+    ) => {
+      metrics.messagesTotal.inc({ direction: 'out' });
+      metrics.bytesTotal.inc({ direction: 'out' }, byteLength(data as Buffer));
+      return (send as (...args: unknown[]) => void)(data, ...rest);
+    }) as typeof ws.send;
+  }
+
+  private startWriting(): void {
+    if (this.editTimer) return;
+    const { editInterval, awarenessInterval } = this.options;
+    // spread the writers over the interval rather than firing them together
+    const first = Math.random() * editInterval;
+    this.editTimer = setTimeout(() => {
+      this.edit();
+      this.editTimer = setInterval(() => this.edit(), editInterval);
+    }, first);
+    if (awarenessInterval > 0) {
+      this.awarenessTimer = setInterval(
+        () => this.awareness(),
+        awarenessInterval,
+      );
+    }
+  }
+
+  private edit(): void {
+    if (this.state !== 'connected') return;
+    this.seq += 1;
+    const text = this.doc.getText(TEXT);
+    const chunk = Array.from(
+      { length: this.options.editSize },
+      () => TYPED[Math.floor(Math.random() * TYPED.length)],
+    ).join('');
+    this.doc.transact(() => {
+      text.insert(text.length, chunk);
+      this.doc
+        .getMap(STAMPS)
+        .set(`w${this.options.id}`, { seq: this.seq, ts: Date.now() });
+    }, this);
+    this.edits += 1;
+    metrics.editsTotal.inc();
+  }
+
+  private awareness(): void {
+    if (this.state !== 'connected' || !this.provider) return;
+    this.provider.awareness.setLocalStateField('cursor', {
+      // what the editor broadcasts, roughly: a position that moves
+      anchor: this.doc.getText(TEXT).length,
+      head: this.doc.getText(TEXT).length,
+      ts: Date.now(),
+    });
+  }
+
+  /** Drop the socket and let the provider reconnect, after `jitter` ms. */
+  storm(jitter: number): void {
+    const provider = this.provider;
+    if (!provider) return;
+    provider.disconnect();
+    this.setState('disconnected');
+    this.synced = false;
+    setTimeout(() => {
+      this.connectAskedAt = Date.now();
+      provider.connect();
+    }, Math.random() * jitter);
+  }
+
+  /** Stop editing; the connection stays, so that the last updates land. */
+  stopWriting(): void {
+    // one handle covers both the initial timeout and the interval that
+    // replaces it: clearing it as both is harmless
+    if (this.editTimer) {
+      clearTimeout(this.editTimer);
+      clearInterval(this.editTimer);
+    }
+    if (this.awarenessTimer) clearInterval(this.awarenessTimer);
+    this.editTimer = null;
+    this.awarenessTimer = null;
+    this.options.writer = false;
+  }
+
+  /** Close for good. */
+  destroy(): void {
+    this.stopWriting();
+    this.provider?.destroy();
+    this.provider = null;
+    this.setState('idle');
+  }
+
+  /** What this client holds, for the convergence check. */
+  fingerprint(): string {
+    const stamps = this.doc.getMap<{ seq: number }>(STAMPS);
+    const entries = [...stamps.entries()]
+      .map(([key, value]) => `${key}=${value.seq}`)
+      .sort()
+      .join(',');
+    return `${this.doc.getText(TEXT).length}|${entries}`;
+  }
+
+  private setState(state: Client['state']): void {
+    if (state === this.state) return;
+    if (this.state !== 'idle')
+      metrics.clientsByState.dec({ state: this.state });
+    this.state = state;
+    if (state !== 'idle') metrics.clientsByState.inc({ state });
+  }
+}
+
+const byteLength = (data: Buffer | ArrayBuffer | Buffer[] | string): number => {
+  if (typeof data === 'string') return Buffer.byteLength(data);
+  if (Array.isArray(data))
+    return data.reduce((sum, chunk) => sum + chunk.byteLength, 0);
+  return data.byteLength;
+};

src/loadtest/swarm/src/config.ts

@@ -0,0 +1,153 @@
+/**
+ * Everything the swarm is told, from the command line, validated once here.
+ *
+ *   swarm --manifest m.json --url wss://docs.example.com --clients 500 \
+ *         --mode wide --duration 300 --ramp 20 --writers 0.3
+ */
+import { parseArgs } from 'node:util';
+
+export type Mode = 'idle' | 'hot' | 'wide';
+
+export interface Config {
+  manifest: string;
+  /** Base of the websocket url, `wss://host` — the room path is appended. */
+  url: string;
+  /** The `Origin` header, which yhub checks against its allowlist. */
+  origin: string;
+  org: string;
+  clients: number;
+  /** Connections opened per second while ramping up. */
+  ramp: number;
+  /** Seconds to hold once every client is connected. */
+  duration: number;
+  mode: Mode;
+  /** `hot`: the one document everybody opens. Defaults to the first public one. */
+  doc?: string;
+  /** Share of the clients that edit, the rest only read. */
+  writers: number;
+  /** Milliseconds between two edits of a writer. */
+  editInterval: number;
+  /** Characters typed per edit. */
+  editSize: number;
+  /** Milliseconds between two awareness updates of a writer, 0 for none. */
+  awarenessInterval: number;
+  /** Second of the hold at which every client disconnects and reconnects, 0 for never. */
+  stormAt: number;
+  /** Maximum jitter, in milliseconds, before a client reconnects in a storm. */
+  reconnectJitter: number;
+  /** Seconds to wait for the documents to converge once the edits stop. */
+  settle: number;
+  metricsPort: number;
+  /** Bearer token the metrics listener requires, empty for none. */
+  metricsToken: string;
+  /** Where the final report is written, `-` for stdout. */
+  report: string;
+  /** Seconds between two progress lines on stderr. */
+  progressInterval: number;
+}
+
+const MODES: Mode[] = ['idle', 'hot', 'wide'];
+
+const number = (
+  name: string,
+  raw: string | undefined,
+  dflt: number,
+  min: number,
+  max = Infinity,
+): number => {
+  if (raw === undefined || raw === '') return dflt;
+  const value = Number(raw);
+  if (!Number.isFinite(value) || value < min || value > max) {
+    throw new Error(
+      `--${name} must be a number between ${min} and ${max} (got "${raw}")`,
+    );
+  }
+  return value;
+};
+
+export const parseConfig = (argv: string[]): Config => {
+  const { values } = parseArgs({
+    args: argv,
+    options: {
+      manifest: { type: 'string' },
+      url: { type: 'string' },
+      origin: { type: 'string' },
+      org: { type: 'string', default: 'docs' },
+      clients: { type: 'string' },
+      ramp: { type: 'string' },
+      duration: { type: 'string' },
+      mode: { type: 'string', default: 'wide' },
+      doc: { type: 'string' },
+      writers: { type: 'string' },
+      'edit-interval': { type: 'string' },
+      'edit-size': { type: 'string' },
+      'awareness-interval': { type: 'string' },
+      'storm-at': { type: 'string' },
+      'reconnect-jitter': { type: 'string' },
+      settle: { type: 'string' },
+      'metrics-port': { type: 'string' },
+      'metrics-token': { type: 'string', default: '' },
+      report: { type: 'string', default: '-' },
+      'progress-interval': { type: 'string' },
+    },
+    strict: true,
+  });
+  if (!values.manifest) throw new Error('--manifest is required');
+  if (!values.url) throw new Error('--url is required (wss://host)');
+  let url: URL;
+  try {
+    url = new URL(values.url);
+  } catch {
+    throw new Error(`--url is not a url (got "${values.url}")`);
+  }
+  if (url.protocol !== 'ws:' && url.protocol !== 'wss:') {
+    throw new Error(`--url must be ws:// or wss:// (got "${values.url}")`);
+  }
+  const mode = values.mode as Mode;
+  if (!MODES.includes(mode)) {
+    throw new Error(
+      `--mode must be one of ${MODES.join(', ')} (got "${values.mode}")`,
+    );
+  }
+  // what a browser on the application would send: the same host, over http(s)
+  const origin =
+    values.origin ??
+    `${url.protocol === 'wss:' ? 'https' : 'http'}://${url.host}`;
+  return {
+    manifest: values.manifest,
+    url: values.url.replace(/\/+$/, ''),
+    origin,
+    org: values.org ?? 'docs',
+    clients: number('clients', values.clients, 10, 1),
+    ramp: number('ramp', values.ramp, 10, 0.1),
+    duration: number('duration', values.duration, 60, 0),
+    mode,
+    doc: values.doc,
+    writers: number('writers', values.writers, mode === 'idle' ? 0 : 0.3, 0, 1),
+    editInterval: number('edit-interval', values['edit-interval'], 2000, 10),
+    editSize: number('edit-size', values['edit-size'], 12, 1, 100000),
+    awarenessInterval: number(
+      'awareness-interval',
+      values['awareness-interval'],
+      5000,
+      0,
+    ),
+    stormAt: number('storm-at', values['storm-at'], 0, 0),
+    reconnectJitter: number(
+      'reconnect-jitter',
+      values['reconnect-jitter'],
+      3000,
+      0,
+    ),
+    settle: number('settle', values.settle, 15, 0),
+    metricsPort: number('metrics-port', values['metrics-port'], 9465, 0, 65535),
+    metricsToken: values['metrics-token'] ?? '',
+    report: values.report ?? '-',
+    progressInterval: number(
+      'progress-interval',
+      values['progress-interval'],
+      10,
+      1,
+    ),
+  };
+};

src/loadtest/swarm/src/index.ts

@@ -0,0 +1,57 @@
+#!/usr/bin/env node
+/**
+ * The command line. Progress goes to stderr, the report to `--report` (stdout
+ * by default), the metrics to `--metrics-port`.
+ */
+import { writeFileSync } from 'node:fs';
+
+import { parseConfig } from './config.js';
+import { readManifest } from './manifest.js';
+import { startMetricsServer } from './metrics.js';
+import { Swarm } from './swarm.js';
+
+const log = (line: string) =>
+  process.stderr.write(`${new Date().toISOString()} ${line}\n`);
+
+const main = async (): Promise<number> => {
+  // y-websocket registers one `exit` listener per provider, and Node warns past
+  // ten of them: thousands are expected here
+  process.setMaxListeners(0);
+  const config = parseConfig(process.argv.slice(2));
+  const manifest = readManifest(config.manifest);
+  if (config.metricsPort > 0) {
+    await startMetricsServer(config.metricsPort, config.metricsToken);
+    log(`metrics on :${config.metricsPort}/metrics`);
+  }
+  const swarm = new Swarm(config, manifest);
+  process.on('SIGINT', () => {
+    log('interrupted: stopping');
+    swarm.stop();
+  });
+  process.on('SIGTERM', () => swarm.stop());
+  log(
+    `starting: ${config.clients} clients, mode ${config.mode}, ramp ${config.ramp}/s, ` +
+      `hold ${config.duration}s, writers ${config.writers}`,
+  );
+  const report = await swarm.run(log);
+  const text = JSON.stringify(report, null, 2);
+  if (config.report === '-') process.stdout.write(`${text}\n`);
+  else writeFileSync(config.report, text);
+  for (const warning of report.warnings) log(`warning: ${warning}`);
+  log(
+    `done: ${report.connected}/${report.clients} connected, ${report.edits} edits, ` +
+      `${report.convergence.converged}/${report.convergence.documents} documents converged`,
+  );
+  return report.convergence.diverged.length === 0 &&
+    report.connected === report.clients
+    ? 0
+    : 1;
+};
+
+main().then(
+  (code) => process.exit(code),
+  (err) => {
+    process.stderr.write(`${err instanceof Error ? err.message : err}\n`);
+    process.exit(2);
+  },
+);

src/loadtest/swarm/src/manifest.ts

@@ -0,0 +1,80 @@
+/**
+ * The manifest `create_load_test_sessions` writes (src/backend/loadtest): which
+ * cookie to send, and which documents each of the logged-in users may open.
+ */
+import { readFileSync } from 'node:fs';
+
+export interface Session {
+  user_id: string;
+  session_key: string;
+  editable_documents: string[];
+  readonly_documents: string[];
+}
+
+export interface Manifest {
+  cookie_name: string;
+  expires_at?: string;
+  public_documents: string[];
+  sessions: Session[];
+}
+
+const isStringArray = (value: unknown): value is string[] =>
+  Array.isArray(value) && value.every((item) => typeof item === 'string');
+
+export const parseManifest = (text: string): Manifest => {
+  let data: unknown;
+  try {
+    data = JSON.parse(text);
+  } catch (err) {
+    throw new Error(
+      `The manifest is not JSON: ${err instanceof Error ? err.message : err}`,
+      { cause: err },
+    );
+  }
+  if (typeof data !== 'object' || data === null) {
+    throw new Error('The manifest is not an object');
+  }
+  const manifest = data as Record<string, unknown>;
+  if (typeof manifest.cookie_name !== 'string' || !manifest.cookie_name) {
+    throw new Error('The manifest has no cookie_name');
+  }
+  if (!Array.isArray(manifest.sessions) || manifest.sessions.length === 0) {
+    throw new Error('The manifest holds no session');
+  }
+  const sessions = manifest.sessions.map((session, index): Session => {
+    const s = session as Record<string, unknown>;
+    if (typeof s.session_key !== 'string' || !s.session_key) {
+      throw new Error(`Session #${index} has no session_key`);
+    }
+    return {
+      user_id: typeof s.user_id === 'string' ? s.user_id : String(index),
+      session_key: s.session_key,
+      editable_documents: isStringArray(s.editable_documents)
+        ? s.editable_documents
+        : [],
+      readonly_documents: isStringArray(s.readonly_documents)
+        ? s.readonly_documents
+        : [],
+    };
+  });
+  if (manifest.expires_at !== undefined) {
+    const expires = Date.parse(String(manifest.expires_at));
+    if (!Number.isNaN(expires) && expires < Date.now()) {
+      throw new Error(
+        `The sessions of the manifest expired at ${manifest.expires_at}`,
+      );
+    }
+  }
+  return {
+    cookie_name: manifest.cookie_name,
+    expires_at:
+      typeof manifest.expires_at === 'string' ? manifest.expires_at : undefined,
+    public_documents: isStringArray(manifest.public_documents)
+      ? manifest.public_documents
+      : [],
+    sessions,
+  };
+};
+
+export const readManifest = (path: string): Manifest =>
+  parseManifest(readFileSync(path, 'utf8'));

src/loadtest/swarm/src/metrics.ts

@@ -0,0 +1,145 @@
+/**
+ * What the swarm measures about itself, served on /metrics for the Prometheus of
+ * the campaign and summarised in the final report.
+ *
+ * Every client-side number the plan asks for is here: time to connect, time to
+ * the first sync, edit propagation latency (a writer stamps the time into the
+ * document, every other client of that document measures the delay when it
+ * arrives), reconnections, close codes, bytes.
+ */
+import { timingSafeEqual } from 'node:crypto';
+import type { IncomingMessage, Server, ServerResponse } from 'node:http';
+import { createServer } from 'node:http';
+
+import {
+  Counter,
+  Gauge,
+  Histogram,
+  Registry,
+  collectDefaultMetrics,
+} from 'prom-client';
+
+export const registry = new Registry();
+
+const LATENCY_BUCKETS = [
+  0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1, 2.5, 5, 10, 30, 60,
+];
+
+export const clientsByState = new Gauge({
+  name: 'swarm_clients',
+  help: 'Virtual clients, by the state of their websocket',
+  labelNames: ['state'] as const,
+  registers: [registry],
+});
+export const connectDuration = new Histogram({
+  name: 'swarm_connect_duration_seconds',
+  help: 'Time from asking for a connection to the socket being open',
+  buckets: LATENCY_BUCKETS,
+  registers: [registry],
+});
+export const syncDuration = new Histogram({
+  name: 'swarm_sync_duration_seconds',
+  help: 'Time from the socket being open to the first sync of the document',
+  buckets: LATENCY_BUCKETS,
+  registers: [registry],
+});
+export const propagationLatency = new Histogram({
+  name: 'swarm_propagation_latency_seconds',
+  help: 'Time for an edit to reach another client of the same document',
+  buckets: LATENCY_BUCKETS,
+  registers: [registry],
+});
+export const editsTotal = new Counter({
+  name: 'swarm_edits_total',
+  help: 'Edits the writers made',
+  registers: [registry],
+});
+export const updatesReceivedTotal = new Counter({
+  name: 'swarm_updates_received_total',
+  help: 'Document updates received from the server',
+  registers: [registry],
+});
+export const messagesTotal = new Counter({
+  name: 'swarm_ws_messages_total',
+  help: 'Websocket messages, by direction',
+  labelNames: ['direction'] as const,
+  registers: [registry],
+});
+export const bytesTotal = new Counter({
+  name: 'swarm_ws_bytes_total',
+  help: 'Websocket payload bytes, by direction',
+  labelNames: ['direction'] as const,
+  registers: [registry],
+});
+export const reconnectsTotal = new Counter({
+  name: 'swarm_reconnects_total',
+  help: 'Connections opened after the first one of a client',
+  registers: [registry],
+});
+export const closesTotal = new Counter({
+  name: 'swarm_ws_closes_total',
+  help: 'Websocket closes, by close code',
+  labelNames: ['code'] as const,
+  registers: [registry],
+});
+export const errorsTotal = new Counter({
+  name: 'swarm_errors_total',
+  help: 'Errors, by kind',
+  labelNames: ['kind'] as const,
+  registers: [registry],
+});
+export const upgradeFailuresTotal = new Counter({
+  name: 'swarm_upgrade_failures_total',
+  help: 'Websocket upgrades refused, by http status',
+  labelNames: ['status'] as const,
+  registers: [registry],
+});
+
+let defaultsCollected = false;
+
+const authorized = (req: IncomingMessage, token: string): boolean => {
+  if (!token) return true;
+  const presented = Buffer.from(req.headers.authorization ?? '');
+  const expected = Buffer.from(`Bearer ${token}`);
+  return (
+    presented.length === expected.length && timingSafeEqual(presented, expected)
+  );
+};
+
+export const handleMetricsRequest = async (
+  req: IncomingMessage,
+  res: ServerResponse,
+  token: string,
+): Promise<void> => {
+  if ((req.url ?? '').split('?')[0] !== '/metrics') {
+    res.writeHead(404).end();
+    return;
+  }
+  if (!authorized(req, token)) {
+    res.writeHead(401, { 'www-authenticate': 'Bearer' }).end();
+    return;
+  }
+  res
+    .writeHead(200, { 'content-type': registry.contentType })
+    .end(await registry.metrics());
+};
+
+export const startMetricsServer = (
+  port: number,
+  token: string,
+): Promise<Server> => {
+  if (!defaultsCollected) {
+    // the event loop lag of the swarm itself: past a point the load generator
+    // is what is saturated, and its numbers stop meaning anything
+    collectDefaultMetrics({ register: registry });
+    defaultsCollected = true;
+  }
+  const server = createServer((req, res) => {
+    void handleMetricsRequest(req, res, token);
+  });
+  server.unref();
+  return new Promise((resolve, reject) => {
+    server.once('error', reject);
+    server.listen(port, () => resolve(server));
+  });
+};

src/loadtest/swarm/src/stats.ts

@@ -0,0 +1,44 @@
+/** Percentiles of what was measured, for the final report. */
+
+export interface Summary {
+  count: number;
+  min: number;
+  p50: number;
+  p95: number;
+  p99: number;
+  max: number;
+  mean: number;
+}
+
+export class Samples {
+  private values: number[] = [];
+
+  add(value: number): void {
+    this.values.push(value);
+  }
+
+  get count(): number {
+    return this.values.length;
+  }
+
+  summary(): Summary | null {
+    if (this.values.length === 0) return null;
+    const sorted = [...this.values].sort((a, b) => a - b);
+    const at = (q: number): number =>
+      sorted[
+        Math.min(
+          sorted.length - 1,
+          Math.max(0, Math.ceil(q * sorted.length) - 1),
+        )
+      ];
+    return {
+      count: sorted.length,
+      min: sorted[0],
+      p50: at(0.5),
+      p95: at(0.95),
+      p99: at(0.99),
+      max: sorted[sorted.length - 1],
+      mean: sorted.reduce((sum, value) => sum + value, 0) / sorted.length,
+    };
+  }
+}

src/loadtest/swarm/src/swarm.ts

@@ -0,0 +1,245 @@
+/**
+ * The run: which client opens which document as whom, the ramp, the hold, the
+ * storm if one is asked for, and the convergence check at the end.
+ */
+import type { Config } from './config.js';
+import { Client } from './client.js';
+import type { Manifest, Session } from './manifest.js';
+import { Samples } from './stats.js';
+import type { Summary } from './stats.js';
+
+export interface Assignment {
+  id: number;
+  session: Session;
+  room: string;
+  writer: boolean;
+}
+
+export interface Report {
+  config: Omit<Config, 'metricsToken'>;
+  startedAt: string;
+  endedAt: string;
+  clients: number;
+  documents: number;
+  writers: number;
+  connected: number;
+  reconnects: number;
+  edits: number;
+  connect: Summary | null;
+  sync: Summary | null;
+  propagation: Summary | null;
+  convergence: {
+    documents: number;
+    converged: number;
+    diverged: string[];
+  };
+  warnings: string[];
+}
+
+/**
+ * Give each virtual client a session and a document.
+ *
+ * Sessions are handed out in turn: with more clients than sessions a user is
+ * logged in from several clients, which the report warns about, the API being
+ * throttled per user. Documents:
+ *   hot   everybody on `config.doc`, else the first public document
+ *   wide  writers open a document they may edit, readers one they may read
+ *         (or edit), picked in turn from their own list
+ *   idle  like wide, and nobody writes
+ */
+export const assign = (
+  config: Config,
+  manifest: Manifest,
+): { assignments: Assignment[]; warnings: string[] } => {
+  const warnings: string[] = [];
+  if (config.clients > manifest.sessions.length) {
+    warnings.push(
+      `${config.clients} clients for ${manifest.sessions.length} sessions: some users are logged in several times, and the API throttles per user`,
+    );
+  }
+  let hotDoc: string | undefined;
+  if (config.mode === 'hot') {
+    hotDoc = config.doc ?? manifest.public_documents[0];
+    if (!hotDoc) {
+      throw new Error(
+        '--mode hot needs --doc, or a public document in the manifest',
+      );
+    }
+  }
+  const writersWanted = Math.round(config.clients * config.writers);
+  const assignments: Assignment[] = [];
+  let skipped = 0;
+  for (let id = 0; assignments.length < config.clients; id++) {
+    if (id >= config.clients + skipped + manifest.sessions.length) break;
+    const session = manifest.sessions[id % manifest.sessions.length];
+    const writer = assignments.filter((a) => a.writer).length < writersWanted;
+    let room: string | undefined;
+    if (hotDoc) {
+      room = hotDoc;
+    } else {
+      const own = writer
+        ? session.editable_documents
+        : [...session.readonly_documents, ...session.editable_documents];
+      const pool = own.length > 0 ? own : manifest.public_documents;
+      room = pool[Math.floor(id / manifest.sessions.length) % pool.length];
+    }
+    if (!room) {
+      skipped += 1;
+      continue;
+    }
+    assignments.push({ id: assignments.length, session, room, writer });
+  }
+  if (assignments.length < config.clients) {
+    warnings.push(
+      `only ${assignments.length} of ${config.clients} clients could be given a document: the manifest lists too few`,
+    );
+  }
+  return { assignments, warnings };
+};
+
+const sleep = (ms: number) => new Promise((resolve) => setTimeout(resolve, ms));
+
+export class Swarm {
+  readonly config: Config;
+  readonly manifest: Manifest;
+  readonly clients: Client[] = [];
+  readonly warnings: string[] = [];
+  private readonly samples = {
+    connect: new Samples(),
+    sync: new Samples(),
+    propagation: new Samples(),
+  };
+  private stopped = false;
+
+  constructor(config: Config, manifest: Manifest) {
+    this.config = config;
+    this.manifest = manifest;
+  }
+
+  /** Ramp up, hold, storm, stop, settle. Resolves with the report. */
+  async run(log: (line: string) => void = () => {}): Promise<Report> {
+    const startedAt = new Date();
+    const { assignments, warnings } = assign(this.config, this.manifest);
+    this.warnings.push(...warnings);
+    const wsBase = `${this.config.url}/collaboration/ws/v1/${this.config.org}`;
+    for (const a of assignments) {
+      this.clients.push(
+        new Client({
+          id: a.id,
+          url: wsBase,
+          origin: this.config.origin,
+          room: a.room,
+          cookie: `${this.manifest.cookie_name}=${a.session.session_key}`,
+          writer: a.writer,
+          editInterval: this.config.editInterval,
+          editSize: this.config.editSize,
+          awarenessInterval: this.config.awarenessInterval,
+          samples: this.samples,
+        }),
+      );
+    }
+
+    const progress = setInterval(
+      () => log(this.progressLine()),
+      this.config.progressInterval * 1000,
+    );
+    try {
+      // ramp: `ramp` connections per second, evenly spaced
+      const interval = 1000 / this.config.ramp;
+      for (const client of this.clients) {
+        if (this.stopped) break;
+        client.connect();
+        await sleep(interval);
+      }
+      log(`ramp done: ${this.clients.length} clients asked to connect`);
+
+      // hold
+      const holdMs = this.config.duration * 1000;
+      const stormMs = this.config.stormAt * 1000;
+      if (stormMs > 0 && stormMs < holdMs) {
+        await this.wait(stormMs);
+        log(
+          `storm: every client disconnects and reconnects within ${this.config.reconnectJitter}ms`,
+        );
+        for (const client of this.clients)
+          client.storm(this.config.reconnectJitter);
+        await this.wait(holdMs - stormMs);
+      } else {
+        await this.wait(holdMs);
+      }
+    } finally {
+      clearInterval(progress);
+    }
+
+    // stop editing, let the last updates land, then compare
+    for (const client of this.clients) client.stopWriting();
+    log(`hold done, settling ${this.config.settle}s`);
+    await sleep(this.config.settle * 1000);
+    const convergence = this.convergence();
+    for (const client of this.clients) client.destroy();
+
+    const { metricsToken: _token, ...config } = this.config;
+    return {
+      config,
+      startedAt: startedAt.toISOString(),
+      endedAt: new Date().toISOString(),
+      clients: this.clients.length,
+      documents: new Set(assignments.map((a) => a.room)).size,
+      writers: assignments.filter((a) => a.writer).length,
+      connected: this.clients.filter((c) => c.connections > 0).length,
+      reconnects: this.clients.reduce(
+        (sum, c) => sum + Math.max(0, c.connections - 1),
+        0,
+      ),
+      edits: this.clients.reduce((sum, c) => sum + c.edits, 0),
+      connect: this.samples.connect.summary(),
+      sync: this.samples.sync.summary(),
+      propagation: this.samples.propagation.summary(),
+      convergence,
+      warnings: this.warnings,
+    };
+  }
+
+  /** Ask the run to end at the next step. */
+  stop(): void {
+    this.stopped = true;
+  }
+
+  private async wait(ms: number): Promise<void> {
+    const until = Date.now() + ms;
+    while (!this.stopped && Date.now() < until) {
+      await sleep(Math.min(250, until - Date.now()));
+    }
+  }
+
+  /** Do all the clients of a document hold the same content? */
+  convergence(): Report['convergence'] {
+    const byRoom = new Map<string, Set<string>>();
+    for (const client of this.clients) {
+      if (client.connections === 0) continue;
+      const fingerprints = byRoom.get(client.options.room) ?? new Set();
+      fingerprints.add(client.fingerprint());
+      byRoom.set(client.options.room, fingerprints);
+    }
+    const diverged = [...byRoom.entries()]
+      .filter(([, f]) => f.size > 1)
+      .map(([room]) => room);
+    return {
+      documents: byRoom.size,
+      converged: byRoom.size - diverged.length,
+      diverged,
+    };
+  }
+
+  progressLine(): string {
+    const counts = { connecting: 0, connected: 0, disconnected: 0, idle: 0 };
+    for (const client of this.clients) counts[client.state] += 1;
+    const propagation = this.samples.propagation.summary();
+    return (
+      `clients connected=${counts.connected} connecting=${counts.connecting} ` +
+      `disconnected=${counts.disconnected} | edits=${this.clients.reduce((s, c) => s + c.edits, 0)} ` +
+      `| propagation p50=${propagation ? (propagation.p50 * 1000).toFixed(0) : '-'}ms ` +
+      `p95=${propagation ? (propagation.p95 * 1000).toFixed(0) : '-'}ms (${propagation?.count ?? 0} samples)`
+    );
+  }
+}

src/loadtest/swarm/tsconfig.build.json

@@ -0,0 +1,9 @@
+{
+  "extends": "./tsconfig.json",
+  "compilerOptions": {
+    "rootDir": "./src",
+    "noEmit": false
+  },
+  "include": ["src/**/*.ts"],
+  "exclude": ["node_modules", "dist", "__tests__"]
+}

src/loadtest/swarm/tsconfig.json

@@ -0,0 +1,24 @@
+{
+  "compilerOptions": {
+    "target": "es2022",
+    "lib": ["es2023"],
+    // the tool runs the compiled JS straight through Node's ESM loader, so
+    // resolution has to be Node's — import specifiers keep their `.js`
+    // extension, as they already did in the JS sources.
+    "module": "nodenext",
+    "moduleResolution": "nodenext",
+    "types": ["node"],
+    "strict": true,
+    "noImplicitOverride": true,
+    "esModuleInterop": true,
+    "skipLibCheck": true,
+    "resolveJsonModule": true,
+    "isolatedModules": true,
+    "verbatimModuleSyntax": true,
+    "noEmit": true,
+    "outDir": "./dist",
+    "sourceMap": true
+  },
+  "include": ["src/**/*.ts", "__tests__/**/*.ts"],
+  "exclude": ["node_modules", "dist"]
+}

src/loadtest/swarm/vitest.config.mts

@@ -0,0 +1,11 @@
+import { defineConfig } from 'vitest/config';
+
+// `__tests__/*.spec.ts`: unit tests of the pure parts, and one integration test
+// running a swarm against an in-process y-websocket server (`_server.ts`).
+export default defineConfig({
+  test: {
+    include: ['__tests__/**/*.spec.ts'],
+    testTimeout: 30000,
+    clearMocks: true,
+  },
+});