Signaling and Perfect Negotiation

{"type":"signal","to":"p_7d4a3f21","data":{"description":{"type":"offer","sdp":"v=0\r\no=- 4611731400430051336 2 IN IP4 127.0.0.1\r\ns=-\r\nt=0 0\r\na=group:BUNDLE 0 1\r\n..."}}}
{"type":"signal","from":"p_deafd35c","data":{"description":{"type":"offer","sdp":"v=0\r\no=- 4611731400430051336 2 IN IP4 127.0.0.1\r\ns=-\r\nt=0 0\r\na=group:BUNDLE 0 1\r\n..."}}}
{"type":"signal","to":"p_7d4a3f21","data":{"candidate":{"candidate":"candidate:1 1 udp 2113937151 0a8d6b0c-0000-4000-8000-0123456789ab.local 54321 typ host","sdpMid":"0","sdpMLineIndex":0,"usernameFragment":"x1Yz"}}}

import { PeerLink } from './perfect-negotiation.js';

// Transport adapters expose the same tiny interface: join(room) -> {id, peers}, signal(to, data), on(event, fn), close().
export async function connectWebSocket(url, ticket) {
  const ws = new WebSocket(`${url}?ticket=${encodeURIComponent(ticket)}`);
  const handlers = new Map();
  const pending = [];
  await new Promise((resolve, reject) => { ws.onopen = resolve; ws.onerror = () => reject(new Error('connect failed')); });
  ws.onmessage = ({ data }) => {
    const msg = JSON.parse(data);
    if (msg.type === 'joined') return pending.shift()?.(msg);
    handlers.get(msg.type)?.(msg);
  };
  return {
    join: (room) => new Promise((resolve) => { pending.push(resolve); ws.send(JSON.stringify({ type: 'join', room })); }),
    signal: (to, data) => ws.send(JSON.stringify({ type: 'signal', to, data })),
    on: (event, fn) => handlers.set(event, fn),
    close: () => ws.close(1000),
  };
}

export async function connectSocketIo(io, ticket) {
  const socket = io({ auth: { ticket }, transports: ['websocket'] });
  await new Promise((resolve, reject) => { socket.on('connect', resolve); socket.on('connect_error', reject); });
  return {
    join: (room) => socket.emitWithAck('join', { room }),
    signal: (to, data) => socket.emit('signal', { to, data }),
    on: (event, fn) => socket.on(event, (payload) => fn(event === 'signal' ? payload : { peer: payload, id: payload.id })),
    close: () => socket.disconnect(),
  };
}

export class Huddle {
  links = new Map();      // peerId -> PeerLink
  streams = new Map();    // peerId -> MediaStream
  #t; #stream; #config; #myId;

  constructor({ transport, stream, config = { iceServers: [] } }) {
    this.#t = transport; this.#stream = stream; this.#config = config;
  }

  async join(room) {
    const t = this.#t;
    t.on('peer-joined', ({ peer }) => this.#link(peer.id));
    t.on('peer-left', ({ id }) => this.#drop(id));
    t.on('signal', ({ from, data }) => this.#link(from).handleSignal(data));
    const joined = await t.join(room);
    this.#myId = joined.id;
    for (const peer of joined.peers) this.#link(peer.id);
    return joined;
  }

  #link(peerId) {
    let link = this.links.get(peerId);
    if (link) return link;
    link = new PeerLink({
      polite: this.#myId > peerId,                   // deterministic roles: both sides compute opposite answers
      config: this.#config,
      send: (data) => this.#t.signal(peerId, data),
      onTrack: ({ streams: [stream] }) => this.streams.set(peerId, stream),
    });
    this.links.set(peerId, link);
    link.addTracks(this.#stream);                    // triggers negotiationneeded on both sides: glare
    return link;
  }

  #drop(peerId) {
    this.links.get(peerId)?.close();
    this.links.delete(peerId);
    this.streams.delete(peerId);
  }

  // Inbound video proves media flows: bytes arrived and frames were decoded.
  async mediaReport() {
    const out = {};
    for (const [peerId, link] of this.links) {
      const report = await link.pc.getStats();
      for (const s of report.values()) {
        if (s.type === 'inbound-rtp' && s.kind === 'video') {
          out[peerId] = { bytesReceived: s.bytesReceived, framesDecoded: s.framesDecoded ?? 0, negotiation: link.stats, state: link.pc.connectionState };
        }
      }
    }
    return out;
  }

  leave() { for (const id of [...this.links.keys()]) this.#drop(id); this.#t.close(); }
}

<!doctype html>
<meta charset="utf-8">
<title>Huddle mesh test</title>
<body>
<script type="module">
import { Huddle, connectWebSocket, connectSocketIo } from '/huddle.js';

// ?kind=ws|socketio&user=alice&room=standup
const params = new URLSearchParams(location.search);
const user = params.get('user') ?? 'guest';

window.startHuddle = async () => {
  const stream = await navigator.mediaDevices.getUserMedia({ audio: true, video: { width: 320, height: 240 } });
  const ticket = await (await fetch(`/ticket?user=${user}`)).text();
  let transport;
  if (params.get('kind') === 'socketio') {
    await import('/socket.io/socket.io.js');               // defines the global io()
    transport = await connectSocketIo(window.io, ticket);
  } else {
    transport = await connectWebSocket(`ws://${location.host}/signal`, ticket);
  }
  window.huddle = new Huddle({ transport, stream });
  return (await window.huddle.join(params.get('room') ?? 'standup')).id;
};
</script>
</body>

import { createServer } from 'node:http';
import { readFile } from 'node:fs/promises';
import { Server } from 'socket.io';
import { issueTicket, redeemTicket } from './ticket.js';

const PORT = Number(process.env.PORT ?? 40402);
const ALLOWED_ORIGINS = (process.env.ALLOWED_ORIGINS ?? `http://localhost:${PORT}`).split(',');
const MAX_ROOM_SIZE = 4;
const RELAY_DELAY_MS = Number(process.env.RELAY_DELAY_MS ?? 0);

const files = { '/': ['index.html', 'text/html'], '/perfect-negotiation.js': ['perfect-negotiation.js', 'text/javascript'], '/huddle.js': ['huddle.js', 'text/javascript'] };

const httpServer = createServer(async (req, res) => {
  const url = new URL(req.url, 'http://localhost');
  if (url.pathname === '/ticket') {
    const user = url.searchParams.get('user') ?? '';
    if (!/^[a-z0-9-]{1,32}$/.test(user)) return res.writeHead(400).end();
    return res.writeHead(200, { 'content-type': 'text/plain', 'cache-control': 'no-store' }).end(issueTicket(user));
  }
  const entry = files[url.pathname];
  if (!entry) return res.writeHead(404).end();
  res.writeHead(200, { 'content-type': entry[1] }).end(await readFile(new URL(`./public/${entry[0]}`, import.meta.url)));
});

const io = new Server(httpServer, {
  path: '/socket.io',
  cors: { origin: ALLOWED_ORIGINS },              // CORS governs the polling transport
  allowRequest: (req, callback) => {              // the WebSocket transport needs an explicit Origin check
    const origin = req.headers.origin;
    callback(null, !origin || ALLOWED_ORIGINS.includes(origin));
  },
  maxHttpBufferSize: 64 * 1024,
});

io.use((socket, next) => {                        // runs once per connection, before 'connection'
  const user = redeemTicket(socket.handshake.auth?.ticket);
  if (!user) return next(new Error('unauthorized'));
  socket.data.name = user;
  next();
});

io.on('connection', (socket) => {
  const peerSummary = (s) => ({ id: s.id, name: s.data.name });

  socket.on('join', async ({ room } = {}, ack) => {
    if (!/^[a-z0-9-]{1,64}$/.test(room ?? '') || typeof ack !== 'function') return;
    const members = await io.in(room).fetchSockets();
    if (members.length >= MAX_ROOM_SIZE) return ack({ ok: false, code: 'room_full' });
    socket.data.room = room;
    await socket.join(room);
    ack({ ok: true, id: socket.id, room, peers: members.map(peerSummary) });
    socket.to(room).emit('peer-joined', peerSummary(socket));
  });

  socket.on('signal', ({ to, data } = {}) => {
    // Relay only inside the sender's room; socket ids are unguessable but never trusted.
    if (!socket.data.room || !io.sockets.adapter.rooms.get(socket.data.room)?.has(to)) return;
    const forward = () => io.to(to).emit('signal', { from: socket.id, data });
    RELAY_DELAY_MS ? setTimeout(forward, RELAY_DELAY_MS) : forward();
  });

  socket.on('disconnecting', () => {
    if (socket.data.room) socket.to(socket.data.room).emit('peer-left', { id: socket.id });
  });
});

httpServer.listen(PORT, () => console.log(`Huddle signalling (Socket.IO) on http://localhost:${PORT}`));
export { httpServer, io };

import { spawn } from 'node:child_process';
import puppeteer from 'puppeteer-core';

const CHROME = process.env.CHROME ?? '/Applications/Google Chrome.app/Contents/MacOS/Google Chrome';
const [kind, port] = process.argv[2] === 'socketio' ? ['socketio', 40402] : ['ws', 40401];
const script = kind === 'socketio' ? 'signaling-socketio.js' : 'signaling-ws.js';

const server = spawn('node', [script], { env: { ...process.env, PORT: String(port), RELAY_DELAY_MS: '150' }, stdio: 'inherit' });
await new Promise((r) => setTimeout(r, 800));

const browser = await puppeteer.launch({
  executablePath: CHROME,
  headless: true,
  args: ['--use-fake-device-for-media-stream', '--use-fake-ui-for-media-stream', '--no-first-run', '--autoplay-policy=no-user-gesture-required'],
});
let failed = false;
try {
  const open = async (user) => {
    const page = await browser.newPage();
    page.on('pageerror', (e) => console.log(`[${user}] pageerror`, e.message));
    await page.goto(`http://localhost:${port}/?kind=${kind}&user=${user}`);
    return page;
  };
  const alice = await open('alice');
  const bob = await open('bob');

  const aliceId = await alice.evaluate(() => window.startHuddle());   // alice joins first and waits
  // Bob joins: both sides add tracks at the same moment and both create an offer.
  const bobId = await bob.evaluate(() => window.startHuddle());
  console.log({ aliceId, bobId });

  const waitMedia = (page) => page.waitForFunction(async () => {
    const r = await window.huddle.mediaReport();
    return Object.values(r).length > 0 && Object.values(r).every((x) => x.bytesReceived > 0 && x.framesDecoded > 0);
  }, { timeout: 20000, polling: 500 });
  await Promise.all([waitMedia(alice), waitMedia(bob)]);

  for (const [name, page] of [['alice', alice], ['bob', bob]]) {
    console.log(name, JSON.stringify(await page.evaluate(() => window.huddle.mediaReport())));
  }
  const stats = (await Promise.all([alice, bob].map((p) => p.evaluate(() => [...window.huddle.links.values()].map((l) => l.stats))))).flat();
  const total = stats.reduce((a, s) => ({ glare: a.glare + s.glare, ignored: a.ignored + s.ignored, rollbacks: a.rollbacks + s.rollbacks }), { glare: 0, ignored: 0, rollbacks: 0 });
  console.log('glare totals', total);
  if (total.glare === 0) { console.log('FAIL: no glare observed'); failed = true; }

} catch (err) {
  console.error('FAIL', err); failed = true;
} finally {
  await browser.close();
  server.kill();
}
process.exit(failed ? 1 : 0);

{
  "name": "huddle-signaling",
  "version": "1.0.0",
  "private": true,
  "type": "module",
  "scripts": {
    "start:ws": "node signaling-ws.js",
    "start:socketio": "node signaling-socketio.js"
  },
  "dependencies": {
    "socket.io": "4.8.4",
    "ws": "8.22.0"
  }
}

npm install puppeteer-core@25.16.0 socket.io-client@4.8.4   # test-only dependencies
node test-two-browsers.js ws          # ws server on port 40401
node test-two-browsers.js socketio    # Socket.IO server on port 40402

{ aliceId: 'p_deafd35c', bobId: 'p_7d4a3f21' }
alice {"p_7d4a3f21":{"bytesReceived":6989,"framesDecoded":3,"negotiation":{"offersSent":1,"answersSent":1,"glare":1,"ignored":0,"rollbacks":1},"state":"connected"}}
bob {"p_deafd35c":{"bytesReceived":10974,"framesDecoded":4,"negotiation":{"offersSent":1,"answersSent":0,"glare":1,"ignored":1,"rollbacks":0},"state":"connected"}}
glare totals { glare: 2, ignored: 1, rollbacks: 1 }

// test-auth.js (excerpt): each attempt resolves to 'open' or 'HTTP <status>'
const attempt = (query, origin) => new Promise((resolve) => {
  const ws = new WebSocket(`ws://localhost:${port}/signal${query}`, { origin });
  ws.on('open', () => { ws.close(); resolve('open'); });
  ws.on('unexpected-response', (_req, res) => resolve(`HTTP ${res.statusCode}`));
  ws.on('error', () => {});
});