2c466e8 · b17dce2 · 8887ee7 · fde810e
@@ -73,3 +73,18 @@
const piSync = piSyncFromEnv(process.env, remote?.http ?? null, remote?.host ??
const relay = await startRelay({ piSync: piSync ?? undefined, host, port, simulation: `http://127.0.0.1:${simulation.port}`, remote, problem, lobby, cameras, leaders, video, summaryEveryMs: 60_000, holdFile: join(dir, "hold.json"), baselineFile: join(dir, "servo-factory.json") });
const { mode, host: armHost } = relay.link();
console.log(`arm on http://${host}:${relay.port}  (${mode === "simulated" ? "simulated" : `arm ${armHost}: ${mode}`}; data in ${dir}${local ? "; local: every visitor drives as the owner" : ""})`);
// A deploy sends SIGTERM: say goodbye on purpose (streams closed, not lost, so the arm server holds the arm; pages told to
// come back at once), flush the lobby's files, and be gone well inside a second.
let stopping = false;
for (const signal of ["SIGTERM", "SIGINT"] as const) {
  process.on(signal, () => {
    if (stopping) return;
    stopping = true;
    setTimeout(() => process.exit(0), 800).unref();
    void relay.close().finally(() => {
      lobby.close();
      process.exit(0);
    });
  });
}
@@ -0,0 +1,77 @@
/**
 * Latency through the whole local pipeline (fake bus, ~1 ms round trip → arm server → relay → page stand-in), per hop:
 *   send→bus      the page sends a frame; the arm server writes Goal_Position (page→relay→Pi→governor→bus)
 *   bus→page      that write until a pushed status shows the joint moved (read-back→Pi→relay→page)
 *   total         command → first confirmed motion at the page
 *   gap           time between pushed statuses while the arm moves (a sawtooth shows as a long, regular gap)
 *   bun bench/pipeline.ts [samples=150]
 */
import { page, startRig } from "../src/pipeline-rig.ts";
const pct = (xs: number[], p: number) => [...xs].sort((a, b) => a - b)[Math.max(0, Math.ceil((p / 100) * xs.length) - 1)] ?? NaN;
const line = (name: string, xs: number[]) => console.log(`${name.padEnd(10)} n=${String(xs.length).padStart(4)}  p50 ${pct(xs, 50).toFixed(1).padStart(6)}  p95 ${pct(xs, 95).toFixed(1).padStart(6)}  max ${Math.max(...xs).toFixed(1).padStart(6)} ms`);
const sleep = (ms: number) => new Promise((r) => setTimeout(r, ms));
const n = Number(process.argv[2] ?? 60);
const rig = await startRig({ latencyMs: Number(process.env.LAT ?? 1) });
const p = await page(rig.base);
let wroteAt = 0;
rig.bus.onSyncWrite = ((prev) => (reg, vals) => {
  if (reg.address === 42 && !wroteAt) wroteAt = performance.now();
  prev?.(reg, vals);
})(rig.bus.onSyncWrite);
p.live.on('close',(c)=>console.log('live closed',c,(performance.now()/1000).toFixed(1)));p.follow.on('close',(c,r)=>console.log('follow closed',c,String(r),(performance.now()/1000).toFixed(1)));
await sleep(600);
let seen: { at: number; pos: number } | null = null;
let last = -1;
let lastAt = 0;
let measuringGaps = false;
const gaps: number[] = [];
p.onStatus((s, at) => {
  const pos = s.joints.find((j) => j.id === 1)!.position;
  seen = { at, pos };
  if (measuringGaps && lastAt) gaps.push(at - lastAt);
  lastAt = at;
});
// Take the arm: first frame eases; wait for it.
let pos = 2048;
p.send({ "1": pos });
await sleep(1500);
const sendBus: number[] = [], busPage: number[] = [], total: number[] = [];
for (let i = 0; i < n; i++) {
  pos += i % 2 ? -24 : 24;
  const m = rig.fake.servos.get(1)!;
  const before = m[56]! | (m[57]! << 8); // ground truth: where the arm is as the command leaves the page
  wroteAt = 0;
  seen = null;
  const t0 = performance.now();
  p.send({ "1": pos });
  const deadline = t0 + 1000;
  while ((!seen || (seen as { pos: number }).pos === before) && performance.now() < deadline) await sleep(0);
  const s = seen as { at: number; pos: number } | null;
  if (s && s.pos !== before && wroteAt) {
    sendBus.push(wroteAt - t0);
    busPage.push(s.at - wroteAt);
    total.push(s.at - t0);
  }
  await sleep(350 + Math.random() * 60);
}
// continuous motion: gaps between statuses
gaps.length = 0;
lastAt = 0;
measuringGaps = true;
for (let i = 0; i < 120; i++) {
  pos += i < 60 ? 12 : -12;
  p.send({ "1": pos });
  await sleep(1000 / 50);
}
console.log(`measured ${total.length} of ${n} commands`);
line("send→bus", sendBus);
line("bus→page", busPage);
line("total", total);
line("gap", gaps);
console.log(`status pushes during 2.4 s of motion: ${gaps.length}  (${((gaps.length / 2.4)).toFixed(1)} Hz)`);
p.close();
await rig.close();
process.exit(0);
@@ -0,0 +1,2 @@
[test]
preload = ["./test-setup.ts"]
@@ -194,6 +194,21 @@
describe("a camera feed", () => {
    await v.cancel();
  });
  test("by default a source that blinked is back in well under a second, not after a fixed two", async () => {
    const closed = createServer();
    const port = await listening(closed);
    closed.close();
    const url = await serveFeed(`http://127.0.0.1:${port}/stream`); // the default retry waits: 200, 500, 1000, 2000 ms
    const v = await watch(url);
    await v.until(hasOffline);
    camera = fakeCamera();
    await new Promise<void>((r) => camera!.server.listen(port, "127.0.0.1", () => r()));
    const up = performance.now();
    await v.until((s) => hasFrame(after(s, PNG)), 3000);
    expect(performance.now() - up).toBeLessThan(450); // the first retry came at about 200 ms
    await v.cancel();
  });
  test("no source: the offline frame at once, and again while anyone watches", async () => {
    const url = await serveFeed(null, { offlineEveryMs: 50 });
    const v = await watch(url);
@@ -19,13 +19,15 @@
export interface FeedOptions {
  connectTimeoutMs?: number;
  /** How long the source may go without a byte before it counts as gone. */
  staleMs?: number;
  /** How soon an unreachable source is tried again, while anyone watches. */
  /** How soon an unreachable source is tried again, while anyone watches: every time, when given. */
  retryMs?: number;
  /** Without `retryMs`: the waits before the 1st, 2nd, 3rd… retry, the last repeating. A source that blinked is back in a fifth of a second. */
  retryBackoffMs?: readonly number[];
  /** How often the offline picture is sent again, so a pane is never left waiting. */
  offlineEveryMs?: number;
  /** How long the upstream connection stays open after the last viewer leaves (a page reload keeps it). */
  lingerMs?: number;
  /** Bytes queued for one viewer past which it skips frames. */
  /** Bytes queued for one viewer past which it skips frames: a few pictures, not a second of them, so a slow link sees the present. */
  maxBacklog?: number;
  /** Where to count frames in from the source and bytes out to viewers (`camera_frames_in`, `camera_bytes_out`). */
  metrics?: LiveMetrics;
@@ -79,7 +81,8 @@
export class CameraFeed {
  private watchdog: ReturnType<typeof setInterval> | undefined;
  private lastByte = 0;
  private closed = false;
  private readonly o: Required<Omit<FeedOptions, "metrics">>;
  private readonly o: Required<Omit<FeedOptions, "metrics" | "retryMs">> & { retryMs?: number };
  private failures = 0;
  private readonly metrics: LiveMetrics | undefined;
  private readonly source: string | null;
  private readonly name: string;
@@ -89,7 +92,7 @@
export class CameraFeed {
    this.source = source;
    this.name = name;
    this.metrics = options.metrics;
    this.o = { connectTimeoutMs: 1500, staleMs: 2000, retryMs: 2000, offlineEveryMs: 2000, lingerMs: 2000, maxBacklog: 1 << 20, ...options };
    this.o = { connectTimeoutMs: 1500, staleMs: 2000, retryBackoffMs: [200, 500, 1000, 2000], offlineEveryMs: 2000, lingerMs: 2000, maxBacklog: 256 << 10, ...options };
    this._state = source ? "idle" : "offline";
  }
@@ -208,7 +211,7 @@
export class CameraFeed {
      this.retry = setTimeout(() => {
        this.retry = undefined;
        if (this.watching && this._state === "offline" && !this.controller) this.connect();
      }, this.o.retryMs);
      }, this.o.retryMs ?? this.o.retryBackoffMs[Math.min(this.failures - 1, this.o.retryBackoffMs.length - 1)] ?? 2000);
      this.retry.unref?.();
    }
  }
@@ -222,6 +225,7 @@
export class CameraFeed {
    this.watchdog = undefined;
    if (this._state !== "offline") console.error(`${this.name} ${this.source}: ${why}; showing it offline`);
    this._state = "offline";
    this.failures++;
    this.broadcastOffline(true);
    this.offline();
  }
@@ -269,6 +273,7 @@
export class CameraFeed {
          if (this._state !== "live") {
            if (this._state === "offline") console.log(`${this.name} ${this.source}: live`);
            this._state = "live";
            this.failures = 0;
            clearInterval(this.repeat);
            this.repeat = undefined;
          }
@@ -0,0 +1,74 @@
import { describe, expect, test } from "bun:test";
import { CommandLag } from "./command-lag.ts";
import { startRig, page } from "./pipeline-rig.ts";
describe("CommandLag", () => {
  test("a command is timed from the arm at rest to the first status whose positions differ", () => {
    const lag = new CommandLag();
    lag.status("1,2", 0);
    lag.status("1,2", 100); // at rest: nothing has changed for 60 ms
    lag.command(110);
    lag.status("1,2", 120); // nothing yet
    lag.status("1,3", 140); // first confirmed motion
    expect(lag.lag(200)).toEqual({ p50: 30, p95: 30, n: 1 });
  });
  test("an arm already moving is not timed, and a command that moves nothing is dropped after the give-up time", () => {
    const lag = new CommandLag(100);
    lag.status("1", 0);
    lag.status("2", 10); // moving
    lag.command(11);
    lag.status("3", 20);
    expect(lag.lag(30)).toBeNull();
    lag.status("3", 100);
    lag.command(101);
    lag.status("3", 300); // too late, and unchanged
    lag.status("4", 320); // a later move is not this command's
    expect(lag.lag(400)).toBeNull();
  });
  test("a second command while one is being timed does not restart the clock; percentiles are over the last samples", () => {
    const lag = new CommandLag();
    lag.status("a", -1000);
    for (let i = 0; i < 20; i++) {
      lag.status("a", i * 1000);
      lag.status("a", i * 1000 + 1);
      lag.command(i * 1000 + 2);
      lag.command(i * 1000 + 6);
      lag.status("b", i * 1000 + 2 + (i >= 18 ? 90 : 10));
      lag.status("b", i * 1000 + 500);
      lag.status("a", i * 1000 + 600);
      lag.status("a", i * 1000 + 900);
    }
    const l = lag.lag(1e9)!;
    expect(l.n).toBe(20);
    expect(l.p50).toBe(10);
    expect(l.p95).toBe(90);
  });
});
describe("through the relay", () => {
  test("the status the page hears carries the measured command-to-motion lag", async () => {
    const rig = await startRig({ latencyMs: 1 });
    const p = await page(rig.base);
    let lag: { p50: number; p95: number; n: number } | undefined;
    p.live.on("message", (d) => {
      const m = JSON.parse(String(d)) as { type?: string; status?: { lag?: typeof lag } | null };
      if (m.type === "status" && m.status?.lag) lag = m.status.lag;
    });
    await new Promise((r) => setTimeout(r, 400));
    let pos = 2048;
    p.send({ "1": pos });
    await new Promise((r) => setTimeout(r, 1500));
    for (let i = 0; i < 12; i++) {
      pos += i % 2 ? -24 : 24;
      p.send({ "1": pos });
      await new Promise((r) => setTimeout(r, 400));
    }
    p.close();
    await rig.close();
    expect(lag).toBeDefined();
    expect(lag!.n).toBeGreaterThanOrEqual(8);
    expect(lag!.p95).toBeGreaterThanOrEqual(lag!.p50); // the numbers themselves are printed by bench/pipeline.ts
  }, 20000);
});
@@ -0,0 +1,69 @@
/**
 * The lag a driver feels: from a command leaving the page to the first confirmed motion of the arm, as the arm's own
 * read-back reports it. Not a round trip: a command is timed only when the arm is at rest (no position has changed for
 * `restMs`), and ends at the first status whose positions differ. A command that moves nothing is dropped
 * after `giveUpMs`. Pure state; the clock is passed in.
 */
export interface Lag {
  p50: number;
  p95: number;
  n: number;
}
const RING = 64;
export class CommandLag {
  private probe: { at: number; base: string } | null = null;
  private current = "";
  private changedAt = -Infinity;
  private readonly samples: number[] = [];
  private cached: Lag | null = null;
  private cachedAt = -Infinity;
  /** Called with every measurement, ms. */
  onSample: ((ms: number) => void) | undefined;
  private readonly giveUpMs: number;
  private readonly restMs: number;
  constructor(giveUpMs = 1000, restMs = 60) {
    this.giveUpMs = giveUpMs;
    this.restMs = restMs;
  }
  /** A command went on to the arm at `now`. */
  command(now: number): void {
    if (this.probe && now - this.probe.at < this.giveUpMs) return;
    this.probe = this.current !== "" && now - this.changedAt >= this.restMs ? { at: now, base: this.current } : null;
  }
  /** A status came in at `now`, with the positions in joint order. */
  status(positions: string, now: number): void {
    if (positions !== this.current) this.changedAt = now;
    this.current = positions;
    const p = this.probe;
    if (!p) return;
    if (now - p.at >= this.giveUpMs) this.probe = null;
    else if (positions !== p.base) {
      this.probe = null;
      this.add(now - p.at);
    }
  }
  private add(ms: number): void {
    if (this.samples.length >= RING) this.samples.shift();
    this.samples.push(ms);
    this.cachedAt = -Infinity;
    this.onSample?.(ms);
  }
  /** p50 and p95 of the last samples (rounded ms); null before there are any. Cached for 500 ms: it is read with every status. */
  lag(now: number): Lag | null {
    if (!this.samples.length) return null;
    if (now - this.cachedAt < 500 && this.cached) return this.cached;
    const sorted = [...this.samples].sort((a, b) => a - b);
    const at = (p: number) => Math.round(sorted[Math.max(0, Math.ceil(p * sorted.length) - 1)]!);
    this.cached = { p50: at(0.5), p95: at(0.95), n: this.samples.length };
    this.cachedAt = now;
    return this.cached;
  }
}
@@ -19,6 +19,8 @@
export interface FakeOptions {
  dropAfter?: Map<number, number>;
  /** Goals are followed at this many ticks a second, read by read, instead of reached at once (bench/precision.ts). */
  speedTicksS?: number;
  /** Every reply comes this many ms after the packet (a real bus round trip is about 1 ms at 1 Mbaud). */
  latencyMs?: number;
}
/** Enough of an STS3215 to answer the bus: a memory map per id, replies by instruction. */
@@ -75,7 +77,7 @@
export class FakeBus implements Transport {
      }
      const delay = this.#options.lateOnce?.get(id) ?? 0;
      this.#options.lateOnce?.delete(id);
      replies.push([delay, frame]);
      replies.push([delay + (this.#options.latencyMs ?? 0), frame]);
    };
    // Instruction packets have the status-packet shape, so the parser reads them too.
    for (const packet of this.#parser.push(data)) {
@@ -0,0 +1,70 @@
import { afterAll, beforeAll, describe, expect, test } from "bun:test";
import { mkdtempSync } from "node:fs";
import { tmpdir } from "node:os";
import { join } from "node:path";
import WebSocket from "ws";
import { FeetechBus } from "./bus.ts";
import { FakeBus } from "./fake-bus.ts";
import { startServer, type RunningServer } from "./server.ts";
const sleep = (ms: number) => new Promise((r) => setTimeout(r, ms));
const IDS = [1, 2, 3];
let fake: FakeBus;
let pi: RunningServer;
beforeAll(async () => {
  fake = new FakeBus(IDS, { latencyMs: 1 });
  pi = await startServer(new FeetechBus(fake), { host: "127.0.0.1", port: 0, ids: IDS, posesFile: join(mkdtempSync(join(tmpdir(), "so101-fb-")), "poses.json") });
});
afterAll(() => Promise.race([pi.close(), sleep(500)]));
/** Statuses a watcher hears for `ms`, with when each came. */
async function listen(ms: number, during?: () => void | Promise<void>) {
  const ws = new WebSocket(`ws://127.0.0.1:${pi.port}/watch`);
  const heard: Array<{ at: number; positions: number[]; torque: boolean[] }> = [];
  ws.on("message", (d) => {
    const m = JSON.parse(String(d)) as { joints?: Array<{ position: number; torque: boolean }> };
    if (m.joints) heard.push({ at: performance.now(), positions: m.joints.map((j) => j.position), torque: m.joints.map((j) => j.torque) });
  });
  await new Promise((r) => ws.once("open", r));
  await sleep(100);
  const from = heard.length;
  const reads = fake.reads;
  await during?.();
  await sleep(ms);
  ws.terminate();
  return { heard: heard.slice(from), reads: fake.reads - reads };
}
const moveJoint = async (steps: number, everyMs: number) => {
  for (let i = 0; i < steps; i++) {
    const m = fake.servos.get(1)!;
    const p = (m[56]! | (m[57]! << 8)) + 3;
    m[56] = p & 0xff;
    m[57] = p >> 8;
    await sleep(everyMs);
  }
};
describe("/watch feedback", () => {
  test("a moving arm is pushed on every change at 25 Hz or better, from one bus read a cycle", async () => {
    const { heard, reads } = await listen(100, () => moveJoint(100, 10));
    // 1.1 s of continuous change: every cycle differs from the last
    expect(heard.length).toBeGreaterThan(40); // 100 changes were made; a 4 a second push would have heard about 5
    // one block read a cycle (the torque read is occasional): well under two bus reads a status
    expect(reads / heard.length).toBeLessThan(1.3);
  });
  test("a quiet arm is only heard from at the heartbeat, not 50 times a second", async () => {
    const { heard } = await listen(1000);
    expect(heard.length).toBeLessThan(40); // not one a cycle: unchanged statuses are held back to the heartbeat
    expect(heard.length).toBeGreaterThanOrEqual(1);
  });
  test("a torque change is in the very next status", async () => {
    const { heard } = await listen(300, async () => {
      await fetch(`http://127.0.0.1:${pi.port}/torque`, { method: "POST", headers: { "content-type": "application/json" }, body: JSON.stringify({ on: true }) });
    });
    expect(heard.some((h) => h.torque.every(Boolean))).toBe(true);
    const first = heard.findIndex((h) => h.torque.every(Boolean));
    expect(heard.slice(first).every((h) => h.torque.every(Boolean))).toBe(true);
  });
});
@@ -0,0 +1,9 @@
import { expect, test } from "bun:test";
import { parseRemote } from "./relay.ts";
test("the suite starts with no arm server or camera from the environment, and a child inherits that", async () => {
  for (const name of Object.keys(process.env)) if (name.startsWith("ARM_")) expect(process.env[name], name).toBe("");
  expect(parseRemote(process.env)).toBeNull();
  const child = Bun.spawn(["bun", "-e", "console.log(JSON.stringify(Object.keys(process.env).filter((k) => k.startsWith('ARM_') && process.env[k])))"], { env: process.env, stdout: "pipe" });
  expect(JSON.parse(await new Response(child.stdout).text())).toEqual([]);
});
@@ -0,0 +1,88 @@
import { describe, expect, test } from "bun:test";
import { DEAD_BYTES, Lobby, SLOW_BYTES, STALL_MS } from "./lobby.ts";
import { behind } from "./send-meter.ts";
/** A page's socket that writes only when told to: sends are held until `flush`, so a slow page costs no bytes and no time. */
function fakePage() {
  const held: Array<{ data: string; done: () => void }> = [];
  const got: string[] = [];
  const ws = {
    readyState: 1,
    terminated: false,
    send(data: string, done?: () => void) {
      held.push({ data, done: done ?? (() => {}) });
    },
    terminate() {
      ws.terminated = true;
      ws.readyState = 3;
    },
    /** The page reads everything held so far. */
    flush() {
      for (const h of held.splice(0)) (got.push(h.data), h.done());
    },
  };
  return { ws, got, held };
}
function room(...pages: Array<ReturnType<typeof fakePage>>) {
  const lobby = new Lobby({ local: true });
  const seats = (lobby as unknown as { seats: Set<unknown> }).seats;
  for (const p of pages) seats.add({ ws: p.ws, who: null, from: {}, verified: false, heard: Date.now() });
  return lobby;
}
const pad = "p".repeat(8000);
const state = (i: number) => JSON.stringify({ type: "x", i, pad });
describe("a page that stops reading", () => {
  test("never stalls the others, is sent only the newest state once it is behind, and gets it when it catches up", () => {
    const fast = fakePage();
    const slow = fakePage();
    const lobby = room(fast, slow);
    const N = 3000;
    let peak = 0;
    for (let i = 0; i < N; i++) {
      lobby.publishState(state(i));
      fast.ws.flush(); // the fast page keeps up
      peak = Math.max(peak, behind(slow.ws));
    }
    expect(fast.got.length).toBe(N); // heard every one
    expect(peak).toBeLessThan(SLOW_BYTES + 16_000); // the slow page's backlog stopped growing at the limit
    expect(slow.held.length).toBeLessThan(20);
    slow.ws.flush(); // it catches up: the newest held state follows at once
    slow.ws.flush();
    expect(JSON.parse(slow.got.at(-1)!).i).toBe(N - 1);
    expect(slow.got.length).toBeLessThan(30); // spared nearly all the steps in between
    expect(slow.ws.terminated).toBe(false);
  });
  test("a page that reads nothing at all is cut once its backlog passes DEAD_BYTES, not buffered without end", () => {
    const dead = fakePage();
    const ok = fakePage();
    const lobby = room(dead, ok);
    const big = JSON.stringify({ type: "chat", pad: "d".repeat(60_000) });
    let sent = 0;
    // chat-style sends reach every page; there is no newest-wins to fall back on
    while (!dead.ws.terminated && sent < 200) {
      lobby.publish(big);
      ok.ws.flush();
      sent++;
    }
    expect(dead.ws.terminated).toBe(true);
    expect(sent).toBeLessThanOrEqual(Math.ceil(DEAD_BYTES / big.length) + 2);
    expect(ok.ws.terminated).toBe(false);
  });
  test("a page whose writes have not moved for STALL_MS is cut at the next push", () => {
    const stuck = fakePage();
    const lobby = room(stuck);
    lobby.publishState(state(0));
    const real = Date.now;
    Date.now = () => real() + STALL_MS + 1000;
    try {
      lobby.publishState(state(1));
      lobby.publishState(state(2));
    } finally {
      Date.now = real;
    }
    expect(stuck.ws.terminated).toBe(true);
  });
});
@@ -38,6 +38,7 @@
 */
import { readFileSync, renameSync, writeFileSync } from "node:fs";
import { behind, meteredSend, stalled } from "./send-meter.ts";
import { randomBytes } from "node:crypto";
import { dirname, join } from "node:path";
import type { IncomingMessage } from "node:http";
@@ -108,8 +109,17 @@
interface Seat {
  guest?: { id: string; name: GuestName };
  /** When this socket's last "who" was answered. */
  whoAt?: number;
  /** The newest state push held back because this socket's send buffer was full: only the latest is kept, and it goes when the buffer drains. */
  latest?: string;
}
/** A page with this much unsent has stopped keeping up: state pushes to it are coalesced into the newest one. */
export const SLOW_BYTES = 64 * 1024;
/** A page with this much unsent is not reading at all: it is cut (its reconnect, with its seat, is cheap) rather than buffered without end. */
/** A page whose writes have not moved for this long is cut: it reconnects, with its seat. */
export const STALL_MS = 8000;
export const DEAD_BYTES = 2 * 1024 * 1024;
const MAX_FRAME = 4096;
/** Lines of chat a new page is sent: enough to fill a phone's chat, little enough for a slow link. */
export const CHAT_FIRST = 30;
@@ -319,6 +329,11 @@
export class Lobby {
    this.wss.handleUpgrade(req, socket, head, (ws) => this.seat(ws, who, from, verified, browserId(req.headers.cookie)));
  }
  /** Tell every page to reconnect (a restart on purpose); each seat's close starts its grace period as for any drop. */
  closeSockets(code: number, reason: string): void {
    for (const s of this.seats) s.ws.close(code, reason);
  }
  close(): void {
    clearInterval(this.pinger);
    clearTimeout(this.presenceTimer);
@@ -355,7 +370,7 @@
export class Lobby {
      this.send(seat, this.adminState());
      this.send(seat, this.crewState());
    }
    for (const listener of this.arriveListeners) listener((text) => seat.ws.readyState === WebSocket.OPEN && seat.ws.send(text));
    for (const listener of this.arriveListeners) listener((text) => this.deliver(seat, text));
    ws.on("message", (data, binary) => {
      seat.heard = Date.now();
      if (binary) return;
@@ -683,7 +698,7 @@
export class Lobby {
  private broadcastAdmin(): void {
    const text = JSON.stringify(this.adminState());
    for (const s of this.seats) if (s.who?.owner && s.ws.readyState === WebSocket.OPEN) s.ws.send(text);
    for (const s of this.seats) if (s.who?.owner) this.deliver(s, text);
  }
  /** What the crew did, for the owners: its settings, the lines it hid, and who it suggests muting. */
@@ -760,14 +775,14 @@
export class Lobby {
  /** Send every open page the same thing, serialized once however many watch; a string is taken as JSON already. */
  publish(body: unknown): void {
    const text = typeof body === "string" ? body : JSON.stringify(body);
    for (const s of this.seats) if (s.ws.readyState === WebSocket.OPEN) s.ws.send(text);
    for (const s of this.seats) this.deliver(s, text);
  }
  /** Tell the driver's pages alone something; in a local run, where nobody takes turns, every page. */
  tellDriver(body: unknown): void {
    const text = JSON.stringify(body);
    const id = this.turns.driver?.id;
    for (const s of this.seats) if (s.ws.readyState === WebSocket.OPEN && (this.local || (id !== undefined && s.who?.id === id))) s.ws.send(text);
    for (const s of this.seats) if (this.local || (id !== undefined && s.who?.id === id)) this.deliver(s, text);
  }
  /** How many pages have the room open. */
@@ -835,11 +850,44 @@
export class Lobby {
  }
  private send(seat: Seat, body: unknown): void {
    if (seat.ws.readyState === WebSocket.OPEN) seat.ws.send(JSON.stringify(body));
    this.deliver(seat, JSON.stringify(body));
  }
  /** One send, the only way text leaves for a page: a socket with a huge or stuck backlog is cut instead of growing without end. */
  private deliver(seat: Seat, text: string): void {
    const ws = seat.ws;
    if (ws.readyState !== WebSocket.OPEN) return;
    if (behind(ws) > DEAD_BYTES || stalled(ws, STALL_MS)) return void ws.terminate();
    meteredSend(ws, text, () => seat.latest !== undefined && behind(ws) <= SLOW_BYTES && this.flush(seat));
  }
  /** The newest held-back state goes to a page that has caught up. */
  private flush(seat: Seat): void {
    const text = seat.latest;
    seat.latest = undefined;
    if (text !== undefined) this.deliver(seat, text);
  }
  /**
   * Push the arm's state: it supersedes the one before, so a page that is behind (more than SLOW_BYTES sent and not yet
   * written out, src/send-meter.ts) is not sent every step, only the newest once it has caught up; everyone else hears
   * every one at once.
   */
  publishState(text: string): void {
    for (const s of this.seats) {
      if (s.ws.readyState !== WebSocket.OPEN) continue;
      if (behind(s.ws) > SLOW_BYTES) {
        s.latest = text;
        if (stalled(s.ws, STALL_MS)) s.ws.terminate();
      } else {
        s.latest = undefined;
        this.deliver(s, text);
      }
    }
  }
  private broadcast(body: unknown): void {
    const text = JSON.stringify(body);
    for (const s of this.seats) if (s.ws.readyState === WebSocket.OPEN) s.ws.send(text);
    for (const s of this.seats) this.deliver(s, text);
  }
}
@@ -128,3 +128,23 @@
test("a panel that opens later gets the log so far; the simulation has a panel t
  first.ws.close();
  later.ws.close();
});
test("a pose that changes at 50 Hz is one log line every 100 ms, and the pose it settles at is logged", async () => {
  const { MachineFeed, NOTE_EVERY_MS } = await import("./machine-feed.ts");
  const feed = new MachineFeed({ link: () => ({ mode: "live", host: null, problem: null, rttMs: null, fps: 0, reconnects: 0 }), room: () => ({ driver: null, paused: false, speedCapPercent: 100 }) });
  const at = (p: number) => ({ joints: [{ id: 1, position: p, torque: true }], driving: null });
  const start = performance.now();
  let p = 2000;
  while (performance.now() - start < 1000) {
    feed.status(at(p++));
    await sleep(20);
  }
  const noted = feed.log.recent(1000).filter((e) => e.what === "status");
  expect(noted.length).toBeLessThanOrEqual(Math.ceil(1000 / NOTE_EVERY_MS) + 1);
  expect(noted.length).toBeGreaterThanOrEqual(7);
  // it stops at p-1; the next status past the interval (a heartbeat) logs where it rests
  await sleep(NOTE_EVERY_MS + 10);
  feed.status(at(p - 1));
  expect((feed.log.recent(1000).filter((e) => e.what === "status").at(-1)!.detail.positions as Record<string, number>)["1"]).toBe(p - 1);
  feed.close();
});
@@ -15,6 +15,7 @@
import type { IncomingMessage } from "node:http";
import type { Duplex } from "node:stream";
import WebSocket, { WebSocketServer } from "ws";
import { behind, meteredSend, stalled } from "./send-meter.ts";
import { MachineLog, perServo, type DetailValue, type EntryKind, type MachineEntry } from "./machine-log.ts";
export interface FeedLink {
@@ -44,6 +45,8 @@
export interface MachineFeedOptions {
/** A viewer with this much unsent is too slow for another status; it still gets the log. */
const SLOW_BYTES = 1 << 20;
const HELLO_ENTRIES = 300;
/** The most often a changed pose is written to the log. */
export const NOTE_EVERY_MS = 100;
type Status = Record<string, unknown>;
interface Joint {
@@ -68,6 +71,7 @@
export class MachineFeed {
  readonly log: MachineLog;
  #status: Status | null = null;
  #signature = "";
  #notedAt = -Infinity;
  #driver: string | null | undefined = undefined;
  /** The arm server's run and the last of its entries kept, so a backlog sent again on reconnecting is not kept twice. */
  #armRun = "";
@@ -105,9 +109,14 @@
export class MachineFeed {
      if (this.#driver !== undefined) this.log.note("state", "turn", { driver: room.driver });
      this.#driver = room.driver;
    }
    // The arm server reads the bus 50 times a second; the log keeps a line for a change at most every NOTE_EVERY_MS, so the
    // minutes of history behind the Machine panel are not pushed out by the poses between. A change held back is noted by the
    // next status past the interval (the quiet arm's heartbeat comes every quarter second), so the pose it rests at is always there.
    const sig = signature(status);
    if (sig !== this.#signature) {
    const now = performance.now();
    if (sig !== this.#signature && now - this.#notedAt >= NOTE_EVERY_MS) {
      this.#signature = sig;
      this.#notedAt = now;
      this.log.note("received", "status", { positions: positions(status), driving: typeof status.driving === "string" ? status.driving : null });
    }
    if (this.#wss.clients.size) this.#send({ type: "status", status, link: this.#o.link(), room }, true);
@@ -140,8 +149,13 @@
export class MachineFeed {
    if (!this.#wss.clients.size) return;
    const text = JSON.stringify(message);
    for (const ws of this.#wss.clients) {
      if (ws.readyState !== WebSocket.OPEN || (droppable && ws.bufferedAmount > SLOW_BYTES)) continue;
      ws.send(text);
      if (ws.readyState !== WebSocket.OPEN) continue;
      if (stalled(ws, 8000)) {
        ws.terminate();
        continue;
      }
      if (droppable && behind(ws) > SLOW_BYTES) continue;
      meteredSend(ws, text);
    }
  }
}
@@ -0,0 +1,73 @@
import { afterEach, describe, expect, test } from "bun:test";
import { mkdtempSync } from "node:fs";
import { tmpdir } from "node:os";
import { join } from "node:path";
import WebSocket from "ws";
import { FeetechBus } from "./bus.ts";
import { FakeBus } from "./fake-bus.ts";
import { startServer, type RunningServer } from "./server.ts";
const sleep = (ms: number) => new Promise((r) => setTimeout(r, ms));
const IDS = [1, 2, 3];
let pi: RunningServer | undefined;
afterEach(() => Promise.race([pi?.close(), sleep(500)]));
async function arm(parkGraceMs: number) {
  const fake = new FakeBus(IDS);
  for (const id of IDS) fake.servos.get(id)!.set([0, 0, 0xff, 0x0f], 9);
  pi = await startServer(new FeetechBus(fake), { host: "127.0.0.1", port: 0, ids: IDS, parkGraceMs, parkSpeedDegS: 2000, posesFile: join(mkdtempSync(join(tmpdir(), "so101-pg-")), "poses.json") } as never);
  return { fake, at: (id: number) => fake.servos.get(id)![56]! | (fake.servos.get(id)![57]! << 8) };
}
const stream = async (port: number) => {
  const ws = new WebSocket(`ws://127.0.0.1:${port}/follow`);
  await new Promise((r) => ws.once("open", r));
  return ws;
};
const drive = async (ws: WebSocket, to: number) => {
  for (let i = 0; i < 20; i++) {
    ws.send(JSON.stringify({ "1": to }));
    await sleep(20);
  }
};
describe("a stream that just stops (the relay restarted) is not an abandoned driver", () => {
  test("the arm holds where it is for the grace, and a stream that returns in time keeps it from parking", async () => {
    const { at } = await arm(600);
    const first = await stream(pi!.port);
    await drive(first, 2600);
    await sleep(300);
    const where = at(1);
    expect(where).toBeGreaterThan(2400);
    first.terminate(); // no close frame: 1006
    await sleep(250);
    expect(at(1)).toBeGreaterThan(where - 40); // holding, not parking
    const second = await stream(pi!.port); // the relay is back with the same driver
    await drive(second, 2600);
    await sleep(900); // past the grace: the first stream's park must have been cancelled
    expect(at(1)).toBeGreaterThan(2400);
    second.close(1000);
  }, 15000);
  test("with no stream back after the grace, the arm parks as before", async () => {
    const { at } = await arm(300);
    const first = await stream(pi!.port);
    await drive(first, 2600);
    await sleep(300);
    expect(at(1)).toBeGreaterThan(2400);
    first.terminate();
    await sleep(200);
    expect(at(1)).toBeGreaterThan(2400);
    for (let i = 0; i < 40 && at(1) > 2300; i++) await sleep(50);
    expect(at(1)).toBeLessThan(2300); // parked toward the middle of its range
  }, 15000);
  test("a driver the relay reports lost (4001) parks at once, grace or not", async () => {
    const { at } = await arm(5000);
    const first = await stream(pi!.port);
    await drive(first, 2600);
    await sleep(300);
    first.close(4001, "the driver's connection was lost");
    for (let i = 0; i < 40 && at(1) > 2300; i++) await sleep(50);
    expect(at(1)).toBeLessThan(2300);
  }, 15000);
});
@@ -0,0 +1,87 @@
/**
 * A whole local pipeline for measuring and testing: a fake bus with a round trip → the arm server (src/server.ts)
 * → the relay (src/relay.ts) → a WebSocket client standing in for the page. Fake arm only; never a real address.
 */
import { mkdtempSync } from "node:fs";
import { tmpdir } from "node:os";
import { join } from "node:path";
import WebSocket from "ws";
import { FeetechBus } from "./bus.ts";
import { stamper } from "./command-gate.ts";
import { FakeBus } from "./fake-bus.ts";
import { Lobby } from "./lobby.ts";
import { parseRemote, startRelay, type RunningRelay } from "./relay.ts";
import { startServer, type RunningServer } from "./server.ts";
import type { ServerOptions } from "./server.ts";
export interface RigOptions {
  /** One bus round trip, ms. */
  latencyMs?: number;
  relay?: Partial<Parameters<typeof startRelay>[0]>;
  server?: Partial<ServerOptions>;
}
export interface Rig {
  fake: FakeBus;
  bus: FeetechBus;
  pi: RunningServer;
  relay: RunningRelay;
  lobby: Lobby;
  dir: string;
  base: string;
  close(): Promise<void>;
}
const IDS = [1, 2, 3, 4, 5, 6];
const sleep = (ms: number) => new Promise((r) => setTimeout(r, ms));
export async function startRig(o: RigOptions = {}): Promise<Rig> {
  const dir = mkdtempSync(join(tmpdir(), "so101-rig-"));
  const fake = new FakeBus(IDS, { latencyMs: o.latencyMs ?? 1 });
  for (const id of IDS) fake.servos.get(id)!.set([0, 0, 0xff, 0x0f], 9); // full range
  const bus = new FeetechBus(fake);
  const pi = await startServer(bus, { host: "127.0.0.1", port: 0, ids: IDS, posesFile: join(dir, "poses.json"), ...o.server });
  const sim = await startServer(new FeetechBus(new FakeBus(IDS)), { host: "127.0.0.1", port: 0, ids: IDS, posesFile: join(dir, "sim-poses.json") });
  const lobby = new Lobby({ local: true });
  const remote = parseRemote({ ARM_SERVER: `ws://127.0.0.1:${pi.port}` })!;
  const relay = await startRelay({ host: "127.0.0.1", port: 0, simulation: `http://127.0.0.1:${sim.port}`, remote, lobby, ...o.relay });
  return {
    fake, bus, pi, relay, lobby, dir,
    base: `127.0.0.1:${relay.port}`,
    close: async () => {
      await Promise.race([relay.close(), sleep(500)]);
      await Promise.race([pi.close(), sleep(500)]);
      await Promise.race([sim.close(), sleep(500)]);
    },
  };
}
export const open = (url: string) =>
  new Promise<WebSocket>((resolve, reject) => {
    const ws = new WebSocket(url);
    ws.once("open", () => resolve(ws));
    ws.once("error", reject);
  });
/** A stand-in for the page: /live for the pushed status, /follow for the stream. */
export async function page(base: string) {
  const live = await open(`ws://${base}/live`);
  const follow = await open(`ws://${base}/follow`);
  const stamp = stamper(() => Date.now());
  const listeners = new Set<(status: { joints: Array<{ id: number; position: number }> }, at: number) => void>();
  live.on("message", (data) => {
    const at = performance.now();
    const text = String(data);
    if (text.startsWith('{"type":"ping"')) return void live.send('{"type":"pong"}');
    if (!text.includes('"status"')) return;
    const m = JSON.parse(text) as { type?: string; status?: { joints: Array<{ id: number; position: number }> } | null };
    if (m.type === "status" && m.status) for (const l of listeners) l(m.status, at);
  });
  live.send(JSON.stringify({ type: "join" }));
  return {
    live, follow,
    send: (joints: Record<string, number>) => follow.send(stamp(joints as never)),
    onStatus: (fn: (typeof listeners extends Set<infer F> ? F : never)) => (listeners.add(fn), () => listeners.delete(fn)),
    close: () => (live.terminate(), follow.terminate()),
  };
}
@@ -0,0 +1,59 @@
import { afterAll, beforeAll, describe, expect, test } from "bun:test";
import { page, startRig, type Rig } from "./pipeline-rig.ts";
// Correctness only: nothing here asserts a time. The latency numbers (p50/p95 per hop) are printed by bench/pipeline.ts.
const sleep = (ms: number) => new Promise((r) => setTimeout(r, ms));
const until = async (done: () => boolean, what: string) => {
  for (let i = 0; i < 3000 && !done(); i++) await sleep(5);
  if (!done()) throw new Error(`never saw ${what}`);
};
let rig: Rig;
let p: Awaited<ReturnType<typeof page>>;
beforeAll(async () => {
  rig = await startRig({ latencyMs: 1 });
  p = await page(rig.base);
  await sleep(300);
});
afterAll(async () => {
  p.close();
  await rig.close();
});
describe("the whole local pipeline (fake bus → arm server → relay → page)", () => {
  test("a move reaches the page as a run of statuses, in order, ending at the target: not one batch per 200 ms", async () => {
    const seen: number[] = [];
    const off = p.onStatus((s) => seen.push(s.joints.find((j) => j.id === 1)!.position));
    let goalWrites = 0;
    const prev = rig.bus.onSyncWrite;
    rig.bus.onSyncWrite = (reg, vals) => {
      if (reg.address === 42) goalWrites++; // Goal_Position
      prev?.(reg, vals);
    };
    p.send({ "1": 2048 });
    await sleep(200);
    seen.length = 0;
    goalWrites = 0;
    const target = 2300;
    let sent = 0;
    // keep the leader's frames coming until the arm has arrived, as a hand does
    await until(() => (p.send({ "1": target }), ++sent, seen.at(-1) === target), "the arm at its target");
    await sleep(100);
    off();
    rig.bus.onSyncWrite = prev;
    const distinct = new Set(seen);
    // in order: a rising move never reads lower than before
    for (let i = 1; i < seen.length; i++) expect(seen[i]!).toBeGreaterThanOrEqual(seen[i - 1]!);
    expect(seen.at(-1)).toBe(target);
    // not batched: the page heard about a fair share of the positions the bus wrote (a 5 a second push would hear one in a dozen)
    expect(goalWrites).toBeGreaterThan(10);
    expect(distinct.size).toBeGreaterThan(goalWrites / 4);
  }, 30000);
  test("a quiet arm still sends statuses (a heartbeat), so the link stays live", async () => {
    let n = 0;
    const off = p.onStatus(() => n++);
    await until(() => n >= 1, "a status from a resting arm");
    off();
  });
});
@@ -62,6 +62,8 @@
import { pipeline } from "node:stream/promises";
import WebSocket, { WebSocketServer } from "ws";
import { LiveMetrics } from "./live-metrics.ts";
import { wallMs } from "./clock.ts";
import { CommandLag } from "./command-lag.ts";
import { behind, meteredSend, stalled } from "./send-meter.ts";
import { ArmLink, type LinkTimings } from "./arm-link.ts";
import { MachineFeed } from "./machine-feed.ts";
import { refusalBatch } from "./refusals.ts";
@@ -186,6 +188,8 @@
const STOPS = new Set(["POST /stop", "POST /estop"]);
const RECONNECTING = "reconnecting to the arm";
/** The close code that tells the arm server its driver was lost, not done: it parks the arm. */
export const LOST_CLOSE = 4001;
/** What a page is told when the relay goes down on purpose: reconnect now. */
export const RESTART_CLOSE = 1012;
const MAX_JOIN_BODY = 1024;
const FLOOR = {
  "x-content-type-options": "nosniff",
@@ -258,6 +262,10 @@
export async function startRelay(options: RelayOptions): Promise<RunningRelay> {
  const lobby = options.lobby ?? null;
  const heartbeatMs = options.heartbeatMs ?? 1000;
  const metrics = new LiveMetrics();
  /** Command → first confirmed motion, timed here from what the arm server reports (src/command-lag.ts). */
  const commandLag = new CommandLag();
  commandLag.onSample = (ms) => metrics.observe("command_to_motion_ms", ms);
  const positionsKey = (status: Record<string, unknown>) => (Array.isArray(status.joints) ? (status.joints as Array<{ position?: unknown }>).map((j) => j.position).join(",") : "");
  const cameras = options.cameras ?? new CameraHub({ feed: { ...options.camera, metrics }, seed: remote ? [{ name: "Camera 1", source: remote.cameras[0]! }] : [] });
  const leaders = options.leaders ?? new LeaderCalibrations();
  const videos = new Map<number, VideoFeed>(Object.entries(options.video ?? {}).map(([id, sources]) => [Number(id), new VideoFeed(sources, {}, options.env ?? process.env)]));
@@ -299,6 +307,7 @@
export async function startRelay(options: RelayOptions): Promise<RunningRelay> {
      metrics,
      onStatus: (status) => {
        if (!onArm()) return;
        commandLag.status(positionsKey(status), performance.now());
        machine.status(status);
        void push(); // on to the pages as it comes, at the arm's 25 Hz: no wait for the next tick
      },
@@ -456,7 +465,8 @@
export async function startRelay(options: RelayOptions): Promise<RunningRelay> {
      const last = armLink!.last;
      const reconnecting = mode() === "reconnecting";
      if (!last) return { code: 503, body: { error: `${RECONNECTING}…`, source: source(), reconnecting } };
      return { code: 200, body: { ...last, held: gate.held ?? last.held ?? null, source: source(), reconnecting, ...(reconnecting ? { seenAt: new Date(armLink!.lastAt).toISOString() } : {}) } };
      const lag = commandLag.lag(performance.now());
      return { code: 200, body: { ...last, held: gate.held ?? last.held ?? null, source: source(), reconnecting, ...(lag ? { lag } : {}), ...(reconnecting ? { seenAt: new Date(armLink!.lastAt).toISOString() } : {}) } };
    }
    const answer = await fetch(simulation + path, { cache: "no-store", signal: AbortSignal.timeout(readTimeoutMs) });
    const body = (await answer.json()) as Record<string, unknown>;
@@ -547,28 +557,29 @@
export async function startRelay(options: RelayOptions): Promise<RunningRelay> {
  const statusEveryMs = options.statusEveryMs ?? 40; // the arm's own bus cycle: the simulation is asked this often
  let pushed = "";
  let pushing = false;
  /** A status came while one was being pushed: push again after, so the newest is never the one left behind. */
  let again = false;
  const push = async (): Promise<void> => {
    if (!lobby) return;
    if (pushing) return void (again = true);
    if (lobby.watching === 0) return void (pushed = ""); // the next page to open the room hears it at once
    // A status that comes in while one is being made is not lost, nor made twice at once: one more pass follows.
    if (pushing) return void (again = true);
    pushing = true;
    let text: string;
    try {
      const { code, body } = await statusOf();
      text = JSON.stringify(code === 200 ? { type: "status", status: body } : { type: "status", status: null, error: body.error ?? `status ${code}` });
    } catch (err) {
      text = JSON.stringify({ type: "status", status: null, error: reason(err) });
      do {
        again = false;
        let text: string;
        try {
          const { code, body } = await statusOf();
          text = JSON.stringify(code === 200 ? { type: "status", status: body } : { type: "status", status: null, error: body.error ?? `status ${code}` });
        } catch (err) {
          text = JSON.stringify({ type: "status", status: null, error: reason(err) });
        }
        if (text !== pushed) lobby.publishState(text);
        pushed = text;
      } while (again);
    } finally {
      pushing = false;
    }
    if (text !== pushed) lobby.publish(text);
    pushed = text;
    if (again) {
      again = false;
      return push();
    }
  };
  /** A static answer, gzipped: whole, so the length is known; kept when the browser may cache it too. */
@@ -932,6 +943,7 @@
export async function startRelay(options: RelayOptions): Promise<RunningRelay> {
        if (lobby) cap.cap(text, percent); // keeps each joint's last target, so a cap set mid-stream starts from there
        machine.note("sent", "follow frame", { values: values(text) });
        arm.send(stamped(data, release));
        if (real) commandLag.command(performance.now());
        return;
      }
      if (!seeded) seed();
@@ -940,6 +952,7 @@
export async function startRelay(options: RelayOptions): Promise<RunningRelay> {
      if (capped === null) return;
      machine.note("sent", "follow frame", { values: values(capped) });
      arm.send(stamped(capped, release));
      if (real) commandLag.command(performance.now());
    });
    arm.on("close", (code, why) => browser.close(sendable(code) ? code : 1011, why.length ? why.toString() : "the arm server went away"));
    // A browser that closed the stream on purpose leaves the arm holding; one that vanished leaves it parking.
@@ -976,7 +989,11 @@
export async function startRelay(options: RelayOptions): Promise<RunningRelay> {
      watched.set(id, (w = { fan, viewers: 0, detach }));
    }
    const mine = w;
    const viewer = { send: (data: Uint8Array) => browser.send(data, () => mine.fan.drain(viewer)), getBufferedAmount: () => browser.bufferedAmount };
    // Counted by src/send-meter.ts, not browser.bufferedAmount (Bun never brings that back down); a viewer stuck for 5 s is cut and reconnects.
    const viewer = {
      send: (data: Uint8Array) => meteredSend(browser, data, () => mine.fan.drain(viewer)),
      getBufferedAmount: () => (stalled(browser, 5000) ? (browser.terminate(), 0) : behind(browser)),
    };
    mine.viewers++;
    mine.fan.add(viewer);
    browser.on("error", () => browser.terminate());
@@ -1092,9 +1109,19 @@
export async function startRelay(options: RelayOptions): Promise<RunningRelay> {
        if (!options.cameras) cameras.close();
        for (const v of videos.values()) v.close();
        for (const c of inFlight) c.abort();
        for (const client of wss.clients) client.terminate();
        // On purpose, so the arm server is told a stream was closed, not lost (it holds the arm, not parks it), and pages are told to come back.
        for (const [browser, arm] of armSides) {
          arm.close(1000);
          browser.close(RESTART_CLOSE, "the relay is restarting");
        }
        lobby?.closeSockets(RESTART_CLOSE, "the relay is restarting");
        const gone = setTimeout(() => {
          for (const client of wss.clients) client.terminate();
          resolve();
        }, 150);
        gone.unref?.();
        wss.close();
        http.close(() => resolve());
        http.close(() => (clearTimeout(gone), resolve()));
        http.closeAllConnections();
      }),
  };
@@ -0,0 +1,176 @@
import { afterAll, describe, expect, test } from "bun:test";
import { mkdtempSync } from "node:fs";
import { tmpdir } from "node:os";
import { join } from "node:path";
import WebSocket from "ws";
import { Accounts } from "./accounts.ts";
import { FeetechBus } from "./bus.ts";
import { stamper } from "./command-gate.ts";
import { FakeBus } from "./fake-bus.ts";
import { sign } from "./identity.ts";
import { Lobby } from "./lobby.ts";
import { Moderator } from "./moderator.ts";
import { parseRemote, startRelay, type RunningRelay } from "./relay.ts";
import { startServer, type RunningServer } from "./server.ts";
const KEY = "restart-test-key";
const sleep = (ms: number) => new Promise((r) => setTimeout(r, ms));
const shut = (s: { close(): Promise<void> } | undefined) => (s ? Promise.race([s.close(), sleep(500)]) : undefined);
const as = (id: string, name: string, role = "user") => {
  const identity = `${id}|${name}|${role}|${Date.now() + 3_600_000}`;
  return { "x-forge-identity": identity, "x-forge-signature": sign(identity, KEY) };
};
type Msg = Record<string, any>;
/** What the app does: its files live in one data directory, and a restart builds everything again from them. */
function app(dir: string, pi: RunningServer, port = 0) {
  const lobby = new Lobby({
    identityKey: KEY,
    chatFile: join(dir, "chat.json"),
    adminFile: join(dir, "admin.json"),
    turnsFile: join(dir, "turns.json"),
    wordsFile: join(dir, "words.json"),
    reportsFile: join(dir, "reports.json"),
    countsFile: join(dir, "visitors.json"),
    accounts: new Accounts({ file: join(dir, "accounts.sqlite") }),
  });
  return startRelay({
    host: "127.0.0.1",
    port,
    simulation: `http://127.0.0.1:${pi.port}`,
    remote: parseRemote({ ARM_SERVER: `ws://127.0.0.1:${pi.port}` })!,
    lobby,
    holdFile: join(dir, "hold.json"),
  }).then((relay) => ({ relay, lobby }));
}
/** The page, as far as a restart is concerned: /live and /follow, each reopened after 100 ms, then doubling. */
class Page {
  heard: Msg[] = [];
  live: WebSocket | null = null;
  follow: WebSocket | null = null;
  stopped = false;
  stamp = stamper(() => Date.now());
  readonly port: () => number;
  readonly headers: Record<string, string>;
  constructor(port: () => number, headers: Record<string, string>) {
    this.port = port;
    this.headers = headers;
    this.openLive(0);
  }
  private openLive(attempt: number) {
    if (this.stopped) return;
    const ws = new WebSocket(`ws://127.0.0.1:${this.port()}/live`, { headers: this.headers });
    this.live = ws;
    let opened = false;
    ws.on("open", () => (opened = true));
    ws.on("message", (d) => {
      const m = JSON.parse(String(d)) as Msg;
      if (m.type === "ping") return ws.send('{"type":"pong"}');
      this.heard.push(m);
    });
    const again = () => {
      if (this.live !== ws) return;
      setTimeout(() => this.openLive(opened ? 1 : attempt + 1), Math.min(100 * 2 ** Math.max(0, (opened ? 1 : attempt + 1) - 1), 2000));
    };
    ws.on("close", again);
    ws.on("error", () => {});
  }
  send(m: Msg) {
    this.live?.readyState === WebSocket.OPEN && this.live.send(JSON.stringify(m));
  }
  driver = () => this.heard.filter((m) => m.type === "state").at(-1)?.driver ?? null;
  state = () => this.heard.filter((m) => m.type === "state").at(-1)!;
  /** Like the page: a frame opens the stream if there is none. */
  command(joints: Record<string, number>) {
    if (!this.follow || this.follow.readyState > 1) {
      const ws = (this.follow = new WebSocket(`ws://127.0.0.1:${this.port()}/follow`, { headers: this.headers }));
      ws.on("error", () => {});
      ws.on("close", () => this.follow === ws && (this.follow = null));
      ws.on("open", () => ws.send(this.stamp(joints)));
    } else if (this.follow.readyState === WebSocket.OPEN) this.follow.send(this.stamp(joints));
  }
  stop() {
    this.stopped = true;
    this.live?.terminate();
    this.follow?.terminate();
  }
}
let pi: RunningServer;
afterAll(() => shut(pi));
describe("a relay restart", () => {
  test("costs under a second: the driver has the turn back with its time, the line and the chat are as they were, and a command reaches the arm", async () => {
    const fake = new FakeBus([1, 2, 3, 4, 5, 6]);
    for (const id of [1, 2, 3, 4, 5, 6]) fake.servos.get(id)!.set([0, 0, 0xff, 0x0f], 9);
    pi = await startServer(new FeetechBus(fake), { host: "127.0.0.1", port: 0, ids: [1, 2, 3, 4, 5, 6], posesFile: join(mkdtempSync(join(tmpdir(), "so101-rs-")), "poses.json") });
    const dir = mkdtempSync(join(tmpdir(), "so101-restart-"));
    let current = await app(dir, pi);
    const port = current.relay.port;
    const alice = new Page(() => port, as("u-alice", "alice"));
    const bob = new Page(() => port, as("u-bob", "bob"));
    const carol = new Page(() => port, as("u-carol", "carol"));
    const owner = new Page(() => port, as("u-owner", "trevato", "admin"));
    await sleep(300);
    owner.send({ type: "mute", id: "u-carol", on: true, for: "forever" });
    owner.send({ type: "settings", set: { turnSeconds: 77 } });
    await sleep(100);
    alice.send({ type: "join" });
    await sleep(100);
    bob.send({ type: "join" });
    await sleep(50);
    carol.send({ type: "join" });
    await sleep(100);
    bob.send({ type: "post", text: "hello from bob" });
    await sleep(100);
    const before = alice.state();
    expect(before.driver.name).toBe("alice");
    expect(before.queue.map((q: Msg) => q.name)).toEqual(["bob", "carol"]);
    alice.command({ "1": 2100 });
    await sleep(600);
    expect(fake.servos.get(1)![56]! | (fake.servos.get(1)![57]! << 8)).toBeGreaterThan(2048);
    // ---- the restart: everything in the process goes, the data directory stays
    const down = performance.now();
    await current.relay.close();
    current.lobby.close();
    const closed = performance.now() - down;
    current = await app(dir, pi, port);
    const started = performance.now() - down;
    console.log(`close ${closed.toFixed(0)} ms, new relay listening ${(started - closed).toFixed(0)} ms after`);
    // the arm was not sent to its park pose by the restart: the relay said goodbye on purpose
    await sleep(100);
    expect(fake.servos.get(1)![56]! | (fake.servos.get(1)![57]! << 8)).toBeGreaterThan(2090);
    // within a second the same driver has the turn back, and a command reaches the arm
    const target = 2300;
    let reached = 0;
    const deadline = down + 1000;
    while (performance.now() < deadline) {
      alice.command({ "1": target });
      await sleep(20);
      const m = fake.servos.get(1)!;
      if ((m[56]! | (m[57]! << 8)) > 2150) {
        reached = performance.now() - down;
        break;
      }
    }
    const after = alice.state();
    console.log(`relay back in ${started.toFixed(0)} ms; first command through ${reached.toFixed(0)} ms after the old one went`);
    await sleep(200); // the other pages are back as well
    expect(reached).toBeGreaterThan(0);
    expect(reached).toBeLessThan(1000);
    expect(after.driver.name).toBe("alice");
    expect(after.driver.endsAt).toBe(before.driver.endsAt); // the same turn, the same remaining time
    expect(after.queue.map((q: Msg) => q.name)).toEqual(["bob", "carol"]);
    expect(bob.heard.filter((m) => m.type === "chat").at(-1)?.messages?.map((m: Msg) => m.text)).toContain("hello from bob");
    expect(carol.heard.filter((m) => m.type === "hello").at(-1)?.muted).toBe(true); // moderation state came back too
    expect(owner.heard.filter((m) => m.type === "admin").at(-1)?.settings.turnSeconds).toBe(77); // and the owner's settings
    owner.stop();
    for (const p of [alice, bob, carol]) p.stop();
    await shut(current.relay);
    current.lobby.close();
  }, 30000);
});
@@ -0,0 +1,32 @@
import { expect, test } from "bun:test";
import { behind, meteredSend, stalled } from "./send-meter.ts";
test("the count is the unconfirmed bytes: it rises with each send and falls with each confirmation", () => {
  const callbacks: Array<() => void> = [];
  const ws = { send: (_d: string | Uint8Array, cb?: () => void) => void callbacks.push(cb!) };
  let drained = 0;
  meteredSend(ws, "x".repeat(100), () => drained++);
  meteredSend(ws, new Uint8Array(50));
  expect(behind(ws)).toBe(150);
  callbacks[0]!();
  expect(behind(ws)).toBe(50);
  expect(drained).toBe(1);
  callbacks[1]!();
  expect(behind(ws)).toBe(0);
  expect(behind({})).toBe(0);
});
test("a socket with bytes waiting and no write confirmed for the limit is stalled; one that is merely slow is not", () => {
  const callbacks: Array<() => void> = [];
  const ws = { send: (_d: string | Uint8Array, cb?: () => void) => void callbacks.push(cb!) };
  const t0 = Date.now();
  expect(stalled(ws, 1000, t0)).toBe(false); // nothing sent
  meteredSend(ws, "a");
  meteredSend(ws, "b");
  expect(stalled(ws, 1000, t0 + 500)).toBe(false);
  expect(stalled(ws, 1000, t0 + 1500)).toBe(true);
  callbacks[0]!(); // progress
  expect(stalled(ws, 1000, Date.now() + 500)).toBe(false);
  callbacks[1]!();
  expect(stalled(ws, 1000, Date.now() + 99_999)).toBe(false); // nothing waiting
});
@@ -0,0 +1,46 @@
/**
 * How far behind a WebSocket is, counted here: bytes handed to `send` whose write has not been confirmed yet.
 *
 * Under Bun, `ws.bufferedAmount` is not to be trusted: once a socket has been backed up it never comes back down, though
 * the bytes have reached the other side (seen with a viewer that stopped reading and then caught up). Anything that used
 * it as "is this viewer keeping up" froze that viewer for good after one slow moment. The write callback does fire when
 * the bytes are really out, so the count is kept from that. A socket that has had unconfirmed bytes and no progress for
 * a long while is `stalled`: the callers cut it, and the page reconnects, rather than wait on a callback that may never come.
 */
interface Sender {
  send(data: string | Uint8Array, callback?: (error?: Error) => void): void;
}
interface Meter {
  bytes: number;
  /** When a write was last confirmed, or when the count left zero. */
  progressAt: number;
}
const meters = new WeakMap<object, Meter>();
/** Unconfirmed bytes on this socket (a string counts its UTF-16 length: close enough for a threshold). */
export function behind(ws: object): number {
  return meters.get(ws)?.bytes ?? 0;
}
/** Whether bytes have been waiting for `ms` with no write confirmed. */
export function stalled(ws: object, ms: number, now = Date.now()): boolean {
  const m = meters.get(ws);
  return m !== undefined && m.bytes > 0 && now - m.progressAt > ms;
}
/** `ws.send`, counted; `done` is called when the write is confirmed (or failed). */
export function meteredSend(ws: Sender, data: string | Uint8Array, done?: () => void): void {
  const size = data.length;
  let m = meters.get(ws);
  if (!m) meters.set(ws, (m = { bytes: 0, progressAt: Date.now() }));
  if (m.bytes === 0) m.progressAt = Date.now();
  m.bytes += size;
  const meter = m;
  ws.send(data, () => {
    meter.bytes = Math.max(0, meter.bytes - size);
    meter.progressAt = Date.now();
    done?.();
  });
}
@@ -50,7 +50,7 @@
 *                           "seq" and "t" (src/command-gate.ts): then a stale, replayed or pre-resume one is dropped.
 *                           A frame over 1 KiB closes the stream; frames past 200 a second are dropped, and a stream that
 *                           keeps that up is closed (src/frame-limit.ts)
 *   WS   /watch             read-only: the server reads the /status body every watchEveryMs (40: the 25 Hz bus cycle), one bus
 *   WS   /watch             read-only: the server reads the /status body every watchEveryMs (17: the 60 Hz bus cycle), one bus
 *                           read shared by every watcher, and sends it when it changed, or after 200 ms of the same;
 *                           the hosted app's relay keeps one open, so viewers see the arm without each of them asking the Pi.
 *                           Between them, {"machine": [entries], "run": "<id>"}: every register written on the bus, every
@@ -121,6 +121,8 @@
export interface ServerOptions {
  now?: () => number;
  /** Longest /watch goes without sending an unchanged status, ms (200): keep it inside the relay link's staleMs. */
  watchQuietMs?: number;
  /** A stream that ends without a close frame (the relay restarted) leaves the arm holding; it parks after this many ms unless a stream returns (default 0 = at once, as for any lost client). */
  parkGraceMs?: number;
}
export interface RunningServer {
@@ -161,6 +163,8 @@
export function actor(body: Record<string, unknown>): string {
const CLEAN_CLOSE = new Set([1000, 1005]);
/** The close code for a stream ended by an emergency stop. */
export const ESTOP_CLOSE = 4423;
/** The close code the relay gives a stream whose driver's browser vanished (src/relay.ts): the arm parks at once, not after the grace. */
const LOST_CLOSE = 4001;
class HttpError extends Error {
  readonly status: number;
@@ -399,6 +403,10 @@
export async function startServer(bus: FeetechBus, options: ServerOptions = {}):
   * The driver's connection is gone: go slowly to the park pose (the saved pose named in the settings, or the middle
   * of every arm joint's range), under supervision, unless the arm is stopped, busy, or the way there is in the zone.
   */
  const parkGraceMs = options.parkGraceMs ?? 0;
  let parkTimer: ReturnType<typeof setTimeout> | undefined;
  /** Streams this server cut itself for not answering pings: a hung client, not a relay that restarted. */
  const heartbeatCut = new WeakSet<WebSocket>();
  const park = async (why: string) => {
    if (estop || gate.held || driver) return;
    const controller = claim("park");
@@ -1175,6 +1183,7 @@
export async function startServer(bus: FeetechBus, options: ServerOptions = {}):
      return;
    }
    claim("stream");
    clearTimeout(parkTimer); // a stream that comes back (the relay restarted) means the arm is not left to park
    const startedAt = Date.now();
    const stats = { frames: 0, dropped: 0, refused: 0, merged: 0, clamped: 0, gated: 0 };
    /** Seq, age and the hold, frame by frame; stamps are optional here (the CLI leader sends none). */
@@ -1305,7 +1314,7 @@
export async function startServer(bus: FeetechBus, options: ServerOptions = {}):
    let answered = true;
    ws.on("pong", () => (answered = true));
    const heartbeat = setInterval(() => {
      if (!answered) return ws.terminate();
      if (!answered) return void (heartbeatCut.add(ws), ws.terminate());
      answered = false;
      ws.ping();
    }, heartbeatMs);
@@ -1524,7 +1533,15 @@
export async function startServer(bus: FeetechBus, options: ServerOptions = {}):
          metrics.event("stream", { seconds: Number(((Date.now() - startedAt) / 1000).toFixed(1)), ...stats, ...(lost ? { lost: code } : {}) });
          release(); // whatever the bus did, the arm is free for the next driver
        }
        if (lost && armed) void park(`the stream was lost (${code})`);
        if (lost && armed) {
          // The relay says its driver was lost with LOST_CLOSE (4001): park now. A stream that just stops (1006: the relay
          // restarted, or the link dropped) is held where it is, and parked only if no stream comes back within the grace.
          if (code === LOST_CLOSE || heartbeatCut.has(ws) || parkGraceMs <= 0) void park(`the stream was lost (${code})`);
          else {
            clearTimeout(parkTimer);
            parkTimer = setTimeout(() => void park(`the stream was lost (${code}) and did not come back in ${parkGraceMs} ms`), parkGraceMs);
          }
        }
      });
    };
  });
@@ -1532,7 +1549,7 @@
export async function startServer(bus: FeetechBus, options: ServerOptions = {}):
  // ------------------------------------------------------------------ the watch
  // Read-only: the status, pushed. One timer reads the bus while anyone watches; a watcher's messages are ignored.
  const watchers = new WebSocketServer({ noServer: true, maxPayload: 1024 });
  const watchEveryMs = options.watchEveryMs ?? 40;
  const watchEveryMs = options.watchEveryMs ?? 17;
  let watching: ReturnType<typeof setInterval> | undefined;
  let reading = false;
  /** The last status sent and when: an unchanged one is not sent again until watchQuietMs pass, so the link stays live. */
@@ -1616,6 +1633,7 @@
export async function startServer(bus: FeetechBus, options: ServerOptions = {}):
    close: () =>
      new Promise<void>((resolve) => {
        clearInterval(watching);
        clearTimeout(parkTimer);
        machine.close();
        precision.close();
        bus.onSyncWrite = null;
@@ -72,6 +72,8 @@
export function createRoom(page) {
  const { $, el, relayed, msg } = page;
  // the type check has no DOM (as ./machine-panel.js)
  const win = /** @type {any} */ (globalThis), doc = win.document, nav = win.navigator, location = win.location;
  /** The newest status pushed and not yet drawn. @type {any} */
  let newest = null;
  /** @type {{ ws: WebSocket | null, hello: any, state: any, mutedList: any[], bannedList: any[], crew?: any, [k: string]: any }} */
  const lobby = { ws: null, hello: null, state: null, offset: 0, retry: 0, attempt: 0, reopening: false, mutedList: [], bannedList: [], admin: null, statusSeen: false };
  function mayDrive() { return !relayed || (lobby.hello !== null && (!lobby.hello.signin || (lobby.hello.you !== null && lobby.state?.driver?.id === lobby.hello.you.id))); }
@@ -357,7 +359,16 @@
export function createRoom(page) {
      if (m.type === "presence") { const t = presenceText(m); if (t !== null) { const b = $("presenceLine"); b.textContent = t; b.disabled = false; } return; }
      if (m.type === "who") return showWho(m);
      if (m.type === "refused") return page.refused?.(m);
      if (m.type === "status") { tell("pose"); lobby.statusSeen = true; return m.status ? page.applyStatus(m.status) : page.noStatus(); }
      if (m.type === "status") {
        tell("pose"); lobby.statusSeen = true;
        if (!m.status) { newest = null; return page.noStatus(); }
        // The relay pushes the arm as fast as the arm server reads it (60 a second): one redraw a frame is enough, and it is the newest.
        const frame = win.requestAnimationFrame?.bind(win);
        if (!frame) return page.applyStatus(m.status);
        if (!newest) frame(() => { const s = newest; newest = null; if (s) page.applyStatus(s); });
        newest = m.status;
        return;
      }
      if (m.type === "hello") lobby.hello = m;
      else if (m.type === "state") { lobby.offset = m.now - Date.now(); lobby.state = m; }
      else if (m.type === "chat") return addChat(m.messages, true), showEarlier(m.more);
@@ -8,6 +8,7 @@
 */
import { proxyFor } from "./egress.ts";
import { behind, meteredSend, stalled } from "./send-meter.ts";
import { AccessUnitSplitter } from "./h264.ts";
import { LADDER, VideoFanout, type FanoutOptions, type VideoSink } from "./video-fanout.ts";
import type { VideoSummary } from "./video-stats.ts";
@@ -19,9 +20,9 @@
export interface VideoSources {
/** The wire the feed talks to a viewer over; a `ws` WebSocket fits. */
export interface VideoSocket {
  bufferedAmount: number;
  readyState: number;
  send(data: Uint8Array | string): void;
  send(data: Uint8Array | string, callback?: (error?: Error) => void): void;
  terminate?(): void;
  on(event: "message", listener: (data: unknown, isBinary: boolean) => void): unknown;
  on(event: "close", listener: () => void): unknown;
}
@@ -52,9 +53,10 @@
export class VideoFeed {
  serve(ws: VideoSocket, driver: boolean): void {
    if (this.closed || !this.available) return;
    const sink: VideoSink = {
      backlog: () => ws.bufferedAmount,
      send: (data) => ws.readyState === OPEN && ws.send(data),
      text: (message) => ws.readyState === OPEN && ws.send(message),
      // Counted by src/send-meter.ts: ws.bufferedAmount never comes back down under Bun, which pinned a viewer to dropping for good.
      backlog: () => (stalled(ws, 5000) ? (ws.terminate?.(), 0) : behind(ws)),
      send: (data) => ws.readyState === OPEN && meteredSend(ws, data),
      text: (message) => ws.readyState === OPEN && meteredSend(ws, message),
    };
    const viewer = this.fanout.join(sink, driver);
    ws.on("message", (data, isBinary) => {
@@ -1559,8 +1559,9 @@
      status = d;
      if ((explainLock() ?? "") !== was) applyLock();
      // The round trip to the arm, which every command also makes: shown once it is long enough to feel.
      const lag = [...rtts].sort((a, b) => a - b)[rtts.length >> 1] ?? 0;
      $("lag").hidden = rtts.length < 3 || lag <= 250; $("lag").textContent = `· ${Math.round(lag)} ms lag`;
      // Command → first confirmed motion, timed by the relay from the arm's own read-back; the camera's round trip only where the relay says nothing.
      const felt = d.lag, lag = felt ? felt.p95 : [...rtts].sort((a, b) => a - b)[rtts.length >> 1] ?? 0;
      $("lag").hidden = felt ? felt.n < 3 || lag <= 100 : rtts.length < 3 || lag <= 250; $("lag").textContent = felt ? `· ${felt.p50}/${felt.p95} ms lag` : `· ${Math.round(lag)} ms lag`;
      $("dot").className = d.reconnecting ? "dot warn" : "dot on";
      $("state").textContent = d.reconnecting ? "reconnecting…" : d.driving === null ? "idle" : d.driving === "record" ? `recording ${d.recording}` : leaderStreaming() ? `teleop · ${teleop.frames} frames` : d.driving;
      if (leaderStreaming()) {
@@ -0,0 +1,7 @@
/**
 * Preloaded by bunfig.toml before every test file: the suite never talks to a real arm. A cell's own app config
 * (ARM_SERVER = the real Pi, its cameras and H.264 sources) is in the environment of the check; every test passes the
 * addresses it means explicitly, and anything it spawns inherits this blank. Blanked, not deleted: Bun keeps reading a
 * deleted variable from where the process started, and the parsers treat a blank one as unset.
 */
for (const name of Object.keys(process.env)) if (name === "ARM_SERVER" || name.startsWith("ARM_CAMERA") || name.startsWith("ARM_H264")) process.env[name] = "";
		@@ -73,3 +73,18 @@ const piSync = piSyncFromEnv(process.env, remote?.http ?? null, remote?.host ??
73	73	const relay = await startRelay({ piSync: piSync ?? undefined, host, port, simulation: `http://127.0.0.1:${simulation.port}`, remote, problem, lobby, cameras, leaders, video, summaryEveryMs: 60_000, holdFile: join(dir, "hold.json"), baselineFile: join(dir, "servo-factory.json") });
74	74	const { mode, host: armHost } = relay.link();
75	75	console.log(`arm on http://${host}:${relay.port} (${mode === "simulated" ? "simulated" : `arm ${armHost}: ${mode}`}; data in ${dir}${local ? "; local: every visitor drives as the owner" : ""})`);
	76
	77	// A deploy sends SIGTERM: say goodbye on purpose (streams closed, not lost, so the arm server holds the arm; pages told to
	78	// come back at once), flush the lobby's files, and be gone well inside a second.
	79	let stopping = false;
	80	for (const signal of ["SIGTERM", "SIGINT"] as const) {
	81	process.on(signal, () => {
	82	if (stopping) return;
	83	stopping = true;
	84	setTimeout(() => process.exit(0), 800).unref();
	85	void relay.close().finally(() => {
	86	lobby.close();
	87	process.exit(0);
	88	});
	89	});
	90	}
		@@ -0,0 +1,77 @@
	1	/**
	2	* Latency through the whole local pipeline (fake bus, ~1 ms round trip → arm server → relay → page stand-in), per hop:
	3	* send→bus the page sends a frame; the arm server writes Goal_Position (page→relay→Pi→governor→bus)
	4	* bus→page that write until a pushed status shows the joint moved (read-back→Pi→relay→page)
	5	* total command → first confirmed motion at the page
	6	* gap time between pushed statuses while the arm moves (a sawtooth shows as a long, regular gap)
	7	* bun bench/pipeline.ts [samples=150]
	8	*/
	9	import { page, startRig } from "../src/pipeline-rig.ts";
	10
	11	const pct = (xs: number[], p: number) => [...xs].sort((a, b) => a - b)[Math.max(0, Math.ceil((p / 100) * xs.length) - 1)] ?? NaN;
	12	const line = (name: string, xs: number[]) => console.log(`${name.padEnd(10)} n=${String(xs.length).padStart(4)} p50 ${pct(xs, 50).toFixed(1).padStart(6)} p95 ${pct(xs, 95).toFixed(1).padStart(6)} max ${Math.max(...xs).toFixed(1).padStart(6)} ms`);
	13	const sleep = (ms: number) => new Promise((r) => setTimeout(r, ms));
	14
	15	const n = Number(process.argv[2] ?? 60);
	16	const rig = await startRig({ latencyMs: Number(process.env.LAT ?? 1) });
	17	const p = await page(rig.base);
	18	let wroteAt = 0;
	19	rig.bus.onSyncWrite = ((prev) => (reg, vals) => {
	20	if (reg.address === 42 && !wroteAt) wroteAt = performance.now();
	21	prev?.(reg, vals);
	22	})(rig.bus.onSyncWrite);
	23	p.live.on('close',(c)=>console.log('live closed',c,(performance.now()/1000).toFixed(1)));p.follow.on('close',(c,r)=>console.log('follow closed',c,String(r),(performance.now()/1000).toFixed(1)));
	24	await sleep(600);
	25
	26	let seen: { at: number; pos: number } \| null = null;
	27	let last = -1;
	28	let lastAt = 0;
	29	let measuringGaps = false;
	30	const gaps: number[] = [];
	31	p.onStatus((s, at) => {
	32	const pos = s.joints.find((j) => j.id === 1)!.position;
	33	seen = { at, pos };
	34	if (measuringGaps && lastAt) gaps.push(at - lastAt);
	35	lastAt = at;
	36	});
	37	// Take the arm: first frame eases; wait for it.
	38	let pos = 2048;
	39	p.send({ "1": pos });
	40	await sleep(1500);
	41	const sendBus: number[] = [], busPage: number[] = [], total: number[] = [];
	42	for (let i = 0; i < n; i++) {
	43	pos += i % 2 ? -24 : 24;
	44	const m = rig.fake.servos.get(1)!;
	45	const before = m[56]! \| (m[57]! << 8); // ground truth: where the arm is as the command leaves the page
	46	wroteAt = 0;
	47	seen = null;
	48	const t0 = performance.now();
	49	p.send({ "1": pos });
	50	const deadline = t0 + 1000;
	51	while ((!seen \|\| (seen as { pos: number }).pos === before) && performance.now() < deadline) await sleep(0);
	52	const s = seen as { at: number; pos: number } \| null;
	53	if (s && s.pos !== before && wroteAt) {
	54	sendBus.push(wroteAt - t0);
	55	busPage.push(s.at - wroteAt);
	56	total.push(s.at - t0);
	57	}
	58	await sleep(350 + Math.random() * 60);
	59	}
	60	// continuous motion: gaps between statuses
	61	gaps.length = 0;
	62	lastAt = 0;
	63	measuringGaps = true;
	64	for (let i = 0; i < 120; i++) {
	65	pos += i < 60 ? 12 : -12;
	66	p.send({ "1": pos });
	67	await sleep(1000 / 50);
	68	}
	69	console.log(`measured ${total.length} of ${n} commands`);
	70	line("send→bus", sendBus);
	71	line("bus→page", busPage);
	72	line("total", total);
	73	line("gap", gaps);
	74	console.log(`status pushes during 2.4 s of motion: ${gaps.length} (${((gaps.length / 2.4)).toFixed(1)} Hz)`);
	75	p.close();
	76	await rig.close();
	77	process.exit(0);
		@@ -194,6 +194,21 @@ describe("a camera feed", () => {
194	194	await v.cancel();
195	195	});
196	196
	197	test("by default a source that blinked is back in well under a second, not after a fixed two", async () => {
	198	const closed = createServer();
	199	const port = await listening(closed);
	200	closed.close();
	201	const url = await serveFeed(`http://127.0.0.1:${port}/stream`); // the default retry waits: 200, 500, 1000, 2000 ms
	202	const v = await watch(url);
	203	await v.until(hasOffline);
	204	camera = fakeCamera();
	205	await new Promise<void>((r) => camera!.server.listen(port, "127.0.0.1", () => r()));
	206	const up = performance.now();
	207	await v.until((s) => hasFrame(after(s, PNG)), 3000);
	208	expect(performance.now() - up).toBeLessThan(450); // the first retry came at about 200 ms
	209	await v.cancel();
	210	});
	211
197	212	test("no source: the offline frame at once, and again while anyone watches", async () => {
198	213	const url = await serveFeed(null, { offlineEveryMs: 50 });
199	214	const v = await watch(url);
		@@ -19,13 +19,15 @@ export interface FeedOptions {
19	19	connectTimeoutMs?: number;
20	20	/** How long the source may go without a byte before it counts as gone. */
21	21	staleMs?: number;
22		/** How soon an unreachable source is tried again, while anyone watches. */
	22	/** How soon an unreachable source is tried again, while anyone watches: every time, when given. */
23	23	retryMs?: number;
	24	/** Without `retryMs`: the waits before the 1st, 2nd, 3rd… retry, the last repeating. A source that blinked is back in a fifth of a second. */
	25	retryBackoffMs?: readonly number[];
24	26	/** How often the offline picture is sent again, so a pane is never left waiting. */
25	27	offlineEveryMs?: number;
26	28	/** How long the upstream connection stays open after the last viewer leaves (a page reload keeps it). */
27	29	lingerMs?: number;
28		/** Bytes queued for one viewer past which it skips frames. */
	30	/** Bytes queued for one viewer past which it skips frames: a few pictures, not a second of them, so a slow link sees the present. */
29	31	maxBacklog?: number;
30	32	/** Where to count frames in from the source and bytes out to viewers (`camera_frames_in`, `camera_bytes_out`). */
31	33	metrics?: LiveMetrics;
		@@ -79,7 +81,8 @@ export class CameraFeed {
79	81	private watchdog: ReturnType<typeof setInterval> \| undefined;
80	82	private lastByte = 0;
81	83	private closed = false;
82		private readonly o: Required<Omit<FeedOptions, "metrics">>;
	84	private readonly o: Required<Omit<FeedOptions, "metrics" \| "retryMs">> & { retryMs?: number };
	85	private failures = 0;
83	86	private readonly metrics: LiveMetrics \| undefined;
84	87	private readonly source: string \| null;
85	88	private readonly name: string;
		@@ -89,7 +92,7 @@ export class CameraFeed {
89	92	this.source = source;
90	93	this.name = name;
91	94	this.metrics = options.metrics;
92		this.o = { connectTimeoutMs: 1500, staleMs: 2000, retryMs: 2000, offlineEveryMs: 2000, lingerMs: 2000, maxBacklog: 1 << 20, ...options };
	95	this.o = { connectTimeoutMs: 1500, staleMs: 2000, retryBackoffMs: [200, 500, 1000, 2000], offlineEveryMs: 2000, lingerMs: 2000, maxBacklog: 256 << 10, ...options };
93	96	this._state = source ? "idle" : "offline";
94	97	}
95	98
		@@ -208,7 +211,7 @@ export class CameraFeed {
208	211	this.retry = setTimeout(() => {
209	212	this.retry = undefined;
210	213	if (this.watching && this._state === "offline" && !this.controller) this.connect();
211		}, this.o.retryMs);
	214	}, this.o.retryMs ?? this.o.retryBackoffMs[Math.min(this.failures - 1, this.o.retryBackoffMs.length - 1)] ?? 2000);
212	215	this.retry.unref?.();
213	216	}
214	217	}
		@@ -222,6 +225,7 @@ export class CameraFeed {
222	225	this.watchdog = undefined;
223	226	if (this._state !== "offline") console.error(`${this.name} ${this.source}: ${why}; showing it offline`);
224	227	this._state = "offline";
	228	this.failures++;
225	229	this.broadcastOffline(true);
226	230	this.offline();
227	231	}
		@@ -269,6 +273,7 @@ export class CameraFeed {
269	273	if (this._state !== "live") {
270	274	if (this._state === "offline") console.log(`${this.name} ${this.source}: live`);
271	275	this._state = "live";
	276	this.failures = 0;
272	277	clearInterval(this.repeat);
273	278	this.repeat = undefined;
274	279	}
		@@ -0,0 +1,74 @@
	1	import { describe, expect, test } from "bun:test";
	2	import { CommandLag } from "./command-lag.ts";
	3	import { startRig, page } from "./pipeline-rig.ts";
	4
	5	describe("CommandLag", () => {
	6	test("a command is timed from the arm at rest to the first status whose positions differ", () => {
	7	const lag = new CommandLag();
	8	lag.status("1,2", 0);
	9	lag.status("1,2", 100); // at rest: nothing has changed for 60 ms
	10	lag.command(110);
	11	lag.status("1,2", 120); // nothing yet
	12	lag.status("1,3", 140); // first confirmed motion
	13	expect(lag.lag(200)).toEqual({ p50: 30, p95: 30, n: 1 });
	14	});
	15
	16	test("an arm already moving is not timed, and a command that moves nothing is dropped after the give-up time", () => {
	17	const lag = new CommandLag(100);
	18	lag.status("1", 0);
	19	lag.status("2", 10); // moving
	20	lag.command(11);
	21	lag.status("3", 20);
	22	expect(lag.lag(30)).toBeNull();
	23	lag.status("3", 100);
	24	lag.command(101);
	25	lag.status("3", 300); // too late, and unchanged
	26	lag.status("4", 320); // a later move is not this command's
	27	expect(lag.lag(400)).toBeNull();
	28	});
	29
	30	test("a second command while one is being timed does not restart the clock; percentiles are over the last samples", () => {
	31	const lag = new CommandLag();
	32	lag.status("a", -1000);
	33	for (let i = 0; i < 20; i++) {
	34	lag.status("a", i * 1000);
	35	lag.status("a", i * 1000 + 1);
	36	lag.command(i * 1000 + 2);
	37	lag.command(i * 1000 + 6);
	38	lag.status("b", i * 1000 + 2 + (i >= 18 ? 90 : 10));
	39	lag.status("b", i * 1000 + 500);
	40	lag.status("a", i * 1000 + 600);
	41	lag.status("a", i * 1000 + 900);
	42	}
	43	const l = lag.lag(1e9)!;
	44	expect(l.n).toBe(20);
	45	expect(l.p50).toBe(10);
	46	expect(l.p95).toBe(90);
	47	});
	48	});
	49
	50	describe("through the relay", () => {
	51	test("the status the page hears carries the measured command-to-motion lag", async () => {
	52	const rig = await startRig({ latencyMs: 1 });
	53	const p = await page(rig.base);
	54	let lag: { p50: number; p95: number; n: number } \| undefined;
	55	p.live.on("message", (d) => {
	56	const m = JSON.parse(String(d)) as { type?: string; status?: { lag?: typeof lag } \| null };
	57	if (m.type === "status" && m.status?.lag) lag = m.status.lag;
	58	});
	59	await new Promise((r) => setTimeout(r, 400));
	60	let pos = 2048;
	61	p.send({ "1": pos });
	62	await new Promise((r) => setTimeout(r, 1500));
	63	for (let i = 0; i < 12; i++) {
	64	pos += i % 2 ? -24 : 24;
	65	p.send({ "1": pos });
	66	await new Promise((r) => setTimeout(r, 400));
	67	}
	68	p.close();
	69	await rig.close();
	70	expect(lag).toBeDefined();
	71	expect(lag!.n).toBeGreaterThanOrEqual(8);
	72	expect(lag!.p95).toBeGreaterThanOrEqual(lag!.p50); // the numbers themselves are printed by bench/pipeline.ts
	73	}, 20000);
	74	});
		@@ -0,0 +1,69 @@
	1	/**
	2	* The lag a driver feels: from a command leaving the page to the first confirmed motion of the arm, as the arm's own
	3	* read-back reports it. Not a round trip: a command is timed only when the arm is at rest (no position has changed for
	4	* `restMs`), and ends at the first status whose positions differ. A command that moves nothing is dropped
	5	* after `giveUpMs`. Pure state; the clock is passed in.
	6	*/
	7	export interface Lag {
	8	p50: number;
	9	p95: number;
	10	n: number;
	11	}
	12
	13	const RING = 64;
	14
	15	export class CommandLag {
	16	private probe: { at: number; base: string } \| null = null;
	17	private current = "";
	18	private changedAt = -Infinity;
	19	private readonly samples: number[] = [];
	20	private cached: Lag \| null = null;
	21	private cachedAt = -Infinity;
	22	/** Called with every measurement, ms. */
	23	onSample: ((ms: number) => void) \| undefined;
	24
	25	private readonly giveUpMs: number;
	26	private readonly restMs: number;
	27
	28	constructor(giveUpMs = 1000, restMs = 60) {
	29	this.giveUpMs = giveUpMs;
	30	this.restMs = restMs;
	31	}
	32
	33	/** A command went on to the arm at `now`. */
	34	command(now: number): void {
	35	if (this.probe && now - this.probe.at < this.giveUpMs) return;
	36	this.probe = this.current !== "" && now - this.changedAt >= this.restMs ? { at: now, base: this.current } : null;
	37	}
	38
	39	/** A status came in at `now`, with the positions in joint order. */
	40	status(positions: string, now: number): void {
	41	if (positions !== this.current) this.changedAt = now;
	42	this.current = positions;
	43	const p = this.probe;
	44	if (!p) return;
	45	if (now - p.at >= this.giveUpMs) this.probe = null;
	46	else if (positions !== p.base) {
	47	this.probe = null;
	48	this.add(now - p.at);
	49	}
	50	}
	51
	52	private add(ms: number): void {
	53	if (this.samples.length >= RING) this.samples.shift();
	54	this.samples.push(ms);
	55	this.cachedAt = -Infinity;
	56	this.onSample?.(ms);
	57	}
	58
	59	/** p50 and p95 of the last samples (rounded ms); null before there are any. Cached for 500 ms: it is read with every status. */
	60	lag(now: number): Lag \| null {
	61	if (!this.samples.length) return null;
	62	if (now - this.cachedAt < 500 && this.cached) return this.cached;
	63	const sorted = [...this.samples].sort((a, b) => a - b);
	64	const at = (p: number) => Math.round(sorted[Math.max(0, Math.ceil(p * sorted.length) - 1)]!);
	65	this.cached = { p50: at(0.5), p95: at(0.95), n: this.samples.length };
	66	this.cachedAt = now;
	67	return this.cached;
	68	}
	69	}
		@@ -19,6 +19,8 @@ export interface FakeOptions {
19	19	dropAfter?: Map<number, number>;
20	20	/** Goals are followed at this many ticks a second, read by read, instead of reached at once (bench/precision.ts). */
21	21	speedTicksS?: number;
	22	/** Every reply comes this many ms after the packet (a real bus round trip is about 1 ms at 1 Mbaud). */
	23	latencyMs?: number;
22	24	}
23	25
24	26	/** Enough of an STS3215 to answer the bus: a memory map per id, replies by instruction. */
		@@ -75,7 +77,7 @@ export class FakeBus implements Transport {
75	77	}
76	78	const delay = this.#options.lateOnce?.get(id) ?? 0;
77	79	this.#options.lateOnce?.delete(id);
78		replies.push([delay, frame]);
	80	replies.push([delay + (this.#options.latencyMs ?? 0), frame]);
79	81	};
80	82	// Instruction packets have the status-packet shape, so the parser reads them too.
81	83	for (const packet of this.#parser.push(data)) {
		@@ -0,0 +1,70 @@
	1	import { afterAll, beforeAll, describe, expect, test } from "bun:test";
	2	import { mkdtempSync } from "node:fs";
	3	import { tmpdir } from "node:os";
	4	import { join } from "node:path";
	5	import WebSocket from "ws";
	6	import { FeetechBus } from "./bus.ts";
	7	import { FakeBus } from "./fake-bus.ts";
	8	import { startServer, type RunningServer } from "./server.ts";
	9
	10	const sleep = (ms: number) => new Promise((r) => setTimeout(r, ms));
	11	const IDS = [1, 2, 3];
	12	let fake: FakeBus;
	13	let pi: RunningServer;
	14	beforeAll(async () => {
	15	fake = new FakeBus(IDS, { latencyMs: 1 });
	16	pi = await startServer(new FeetechBus(fake), { host: "127.0.0.1", port: 0, ids: IDS, posesFile: join(mkdtempSync(join(tmpdir(), "so101-fb-")), "poses.json") });
	17	});
	18	afterAll(() => Promise.race([pi.close(), sleep(500)]));
	19
	20	/** Statuses a watcher hears for `ms`, with when each came. */
	21	async function listen(ms: number, during?: () => void \| Promise<void>) {
	22	const ws = new WebSocket(`ws://127.0.0.1:${pi.port}/watch`);
	23	const heard: Array<{ at: number; positions: number[]; torque: boolean[] }> = [];
	24	ws.on("message", (d) => {
	25	const m = JSON.parse(String(d)) as { joints?: Array<{ position: number; torque: boolean }> };
	26	if (m.joints) heard.push({ at: performance.now(), positions: m.joints.map((j) => j.position), torque: m.joints.map((j) => j.torque) });
	27	});
	28	await new Promise((r) => ws.once("open", r));
	29	await sleep(100);
	30	const from = heard.length;
	31	const reads = fake.reads;
	32	await during?.();
	33	await sleep(ms);
	34	ws.terminate();
	35	return { heard: heard.slice(from), reads: fake.reads - reads };
	36	}
	37	const moveJoint = async (steps: number, everyMs: number) => {
	38	for (let i = 0; i < steps; i++) {
	39	const m = fake.servos.get(1)!;
	40	const p = (m[56]! \| (m[57]! << 8)) + 3;
	41	m[56] = p & 0xff;
	42	m[57] = p >> 8;
	43	await sleep(everyMs);
	44	}
	45	};
	46
	47	describe("/watch feedback", () => {
	48	test("a moving arm is pushed on every change at 25 Hz or better, from one bus read a cycle", async () => {
	49	const { heard, reads } = await listen(100, () => moveJoint(100, 10));
	50	// 1.1 s of continuous change: every cycle differs from the last
	51	expect(heard.length).toBeGreaterThan(40); // 100 changes were made; a 4 a second push would have heard about 5
	52	// one block read a cycle (the torque read is occasional): well under two bus reads a status
	53	expect(reads / heard.length).toBeLessThan(1.3);
	54	});
	55
	56	test("a quiet arm is only heard from at the heartbeat, not 50 times a second", async () => {
	57	const { heard } = await listen(1000);
	58	expect(heard.length).toBeLessThan(40); // not one a cycle: unchanged statuses are held back to the heartbeat
	59	expect(heard.length).toBeGreaterThanOrEqual(1);
	60	});
	61
	62	test("a torque change is in the very next status", async () => {
	63	const { heard } = await listen(300, async () => {
	64	await fetch(`http://127.0.0.1:${pi.port}/torque`, { method: "POST", headers: { "content-type": "application/json" }, body: JSON.stringify({ on: true }) });
	65	});
	66	expect(heard.some((h) => h.torque.every(Boolean))).toBe(true);
	67	const first = heard.findIndex((h) => h.torque.every(Boolean));
	68	expect(heard.slice(first).every((h) => h.torque.every(Boolean))).toBe(true);
	69	});
	70	});
		@@ -0,0 +1,9 @@
	1	import { expect, test } from "bun:test";
	2	import { parseRemote } from "./relay.ts";
	3
	4	test("the suite starts with no arm server or camera from the environment, and a child inherits that", async () => {
	5	for (const name of Object.keys(process.env)) if (name.startsWith("ARM_")) expect(process.env[name], name).toBe("");
	6	expect(parseRemote(process.env)).toBeNull();
	7	const child = Bun.spawn(["bun", "-e", "console.log(JSON.stringify(Object.keys(process.env).filter((k) => k.startsWith('ARM_') && process.env[k])))"], { env: process.env, stdout: "pipe" });
	8	expect(JSON.parse(await new Response(child.stdout).text())).toEqual([]);
	9	});
		@@ -0,0 +1,88 @@
	1	import { describe, expect, test } from "bun:test";
	2	import { DEAD_BYTES, Lobby, SLOW_BYTES, STALL_MS } from "./lobby.ts";
	3	import { behind } from "./send-meter.ts";
	4
	5	/** A page's socket that writes only when told to: sends are held until `flush`, so a slow page costs no bytes and no time. */
	6	function fakePage() {
	7	const held: Array<{ data: string; done: () => void }> = [];
	8	const got: string[] = [];
	9	const ws = {
	10	readyState: 1,
	11	terminated: false,
	12	send(data: string, done?: () => void) {
	13	held.push({ data, done: done ?? (() => {}) });
	14	},
	15	terminate() {
	16	ws.terminated = true;
	17	ws.readyState = 3;
	18	},
	19	/** The page reads everything held so far. */
	20	flush() {
	21	for (const h of held.splice(0)) (got.push(h.data), h.done());
	22	},
	23	};
	24	return { ws, got, held };
	25	}
	26	function room(...pages: Array<ReturnType<typeof fakePage>>) {
	27	const lobby = new Lobby({ local: true });
	28	const seats = (lobby as unknown as { seats: Set<unknown> }).seats;
	29	for (const p of pages) seats.add({ ws: p.ws, who: null, from: {}, verified: false, heard: Date.now() });
	30	return lobby;
	31	}
	32	const pad = "p".repeat(8000);
	33	const state = (i: number) => JSON.stringify({ type: "x", i, pad });
	34
	35	describe("a page that stops reading", () => {
	36	test("never stalls the others, is sent only the newest state once it is behind, and gets it when it catches up", () => {
	37	const fast = fakePage();
	38	const slow = fakePage();
	39	const lobby = room(fast, slow);
	40	const N = 3000;
	41	let peak = 0;
	42	for (let i = 0; i < N; i++) {
	43	lobby.publishState(state(i));
	44	fast.ws.flush(); // the fast page keeps up
	45	peak = Math.max(peak, behind(slow.ws));
	46	}
	47	expect(fast.got.length).toBe(N); // heard every one
	48	expect(peak).toBeLessThan(SLOW_BYTES + 16_000); // the slow page's backlog stopped growing at the limit
	49	expect(slow.held.length).toBeLessThan(20);
	50	slow.ws.flush(); // it catches up: the newest held state follows at once
	51	slow.ws.flush();
	52	expect(JSON.parse(slow.got.at(-1)!).i).toBe(N - 1);
	53	expect(slow.got.length).toBeLessThan(30); // spared nearly all the steps in between
	54	expect(slow.ws.terminated).toBe(false);
	55	});
	56
	57	test("a page that reads nothing at all is cut once its backlog passes DEAD_BYTES, not buffered without end", () => {
	58	const dead = fakePage();
	59	const ok = fakePage();
	60	const lobby = room(dead, ok);
	61	const big = JSON.stringify({ type: "chat", pad: "d".repeat(60_000) });
	62	let sent = 0;
	63	// chat-style sends reach every page; there is no newest-wins to fall back on
	64	while (!dead.ws.terminated && sent < 200) {
	65	lobby.publish(big);
	66	ok.ws.flush();
	67	sent++;
	68	}
	69	expect(dead.ws.terminated).toBe(true);
	70	expect(sent).toBeLessThanOrEqual(Math.ceil(DEAD_BYTES / big.length) + 2);
	71	expect(ok.ws.terminated).toBe(false);
	72	});
	73
	74	test("a page whose writes have not moved for STALL_MS is cut at the next push", () => {
	75	const stuck = fakePage();
	76	const lobby = room(stuck);
	77	lobby.publishState(state(0));
	78	const real = Date.now;
	79	Date.now = () => real() + STALL_MS + 1000;
	80	try {
	81	lobby.publishState(state(1));
	82	lobby.publishState(state(2));
	83	} finally {
	84	Date.now = real;
	85	}
	86	expect(stuck.ws.terminated).toBe(true);
	87	});
	88	});