Public: anyone can read this; the crew can change it.
Migrated from github.com/Trevato/SO-ARM100
workbenchthis machine
When a turn ends, the relay fetches its trace from GET /precision/trace and stores it with the turn in facts (capped, gone at 90 days) #552
smith/arm:smith/547-when-a-turn-ends-the-relay-fetches-its-t → main 1 commit fork
4 files
changed +132
−8
Software/so101/src/facts.test.ts +75 −1
| @@ -3,7 +3,7 @@ import { mkdtempSync } from "node:fs"; | ||
| 3 | 3 | import { tmpdir } from "node:os"; |
| 4 | 4 | import { join } from "node:path"; |
| 5 | 5 | import { ArmEdges } from "./arm-edges.ts"; |
| 6 | import { Facts, KEEP_MS, MAX_VISITS_PER_HOUR, parseCard, whoIs } from "./facts.ts"; | |
| 6 | import { Facts, KEEP_MS, MAX_TRACE_BYTES, MAX_VISITS_PER_HOUR, parseCard, parseSessionId, whoIs } from "./facts.ts"; | |
| 7 | 7 | import type { Snapshot } from "./turns.ts"; |
| 8 | 8 | |
| 9 | 9 | const person = (id: string, name = id) => ({ id, name, owner: false }); |
| @@ -161,3 +161,77 @@ test("the relay writes app start and a turn's row with its card", async () => { | ||
| 161 | 161 | await stop(); |
| 162 | 162 | } |
| 163 | 163 | }); |
| 164 | ||
| 165 | const trace = (n: number) => ({ session: 7, t0: 1, hz: 5, joints: { shoulder_pan: { intent: Array(n).fill(2048), actual: Array(n).fill(2000) } } }); | |
| 166 | ||
| 167 | test("a trace round-trips with its turn, is capped, and goes with the turn at 90 days", () => { | |
| 168 | let t = 1000; | |
| 169 | const facts = new Facts({ commit: "t", now: () => t }); | |
| 170 | facts.turn({ kind: "turn", ended: null, started: person("a") } as never, snap()); | |
| 171 | const id = facts.turn({ kind: "turn", ended: { who: person("a"), reason: "done" }, started: null } as never, snap())!; | |
| 172 | expect(facts.traceFor(id)).toBeNull(); | |
| 173 | facts.trace(id, trace(3)); | |
| 174 | expect(facts.traceFor(id)).toEqual(trace(3)); | |
| 175 | facts.trace(id, "not a trace"); | |
| 176 | expect(facts.traceFor(id)).toEqual(trace(3)); | |
| 177 | facts.trace(id, trace(200_000)); | |
| 178 | const big = facts.traceFor(id) as ReturnType<typeof trace> & { truncated: boolean }; | |
| 179 | expect(JSON.stringify(big).length).toBeLessThanOrEqual(MAX_TRACE_BYTES); | |
| 180 | expect(big.truncated).toBe(true); | |
| 181 | expect(big.joints.shoulder_pan.intent.length).toBe(big.joints.shoulder_pan.actual.length); | |
| 182 | t = 1000 + KEEP_MS + 1; | |
| 183 | facts.prune(); | |
| 184 | expect(facts.traceFor(id)).toBeNull(); | |
| 185 | }); | |
| 186 | ||
| 187 | test("the session of a turn is the newest stored one that ended since it began", () => { | |
| 188 | const at = new Date(5000).toISOString(); | |
| 189 | expect(parseSessionId({ sessions: [{ id: 3, at }] }, 4000)).toBe(3); | |
| 190 | expect(parseSessionId({ sessions: [{ id: 3, at }] }, 6000)).toBeNull(); | |
| 191 | expect(parseSessionId({ sessions: [{ at }] }, 0)).toBeNull(); | |
| 192 | expect(parseSessionId({}, 0)).toBeNull(); | |
| 193 | }); | |
| 194 | ||
| 195 | async function traceRelay(route: "trace" | "404") { | |
| 196 | const facts = new Facts({ commit: "t" }); | |
| 197 | const { Lobby } = await import("./lobby.ts"); | |
| 198 | const { startRelay } = await import("./relay.ts"); | |
| 199 | const sim = Bun.serve({ | |
| 200 | port: 0, | |
| 201 | hostname: "127.0.0.1", | |
| 202 | fetch: (req) => { | |
| 203 | const path = new URL(req.url).pathname; | |
| 204 | if (path === "/status") return Response.json({ joints: [], held: null }); | |
| 205 | if (path === "/precision") return Response.json({ session: { releases: 1 }, sessions: [{ id: 9, at: new Date(Date.now() + 5000).toISOString() }] }); | |
| 206 | return route === "trace" ? Response.json(trace(4)) : Response.json({ error: "no" }, { status: 404 }); | |
| 207 | }, | |
| 208 | }); | |
| 209 | const lobby = new Lobby({ identityKey: "k".repeat(64) }); | |
| 210 | const relay = await startRelay({ port: 0, simulation: `http://127.0.0.1:${sim.port}`, lobby, facts }); | |
| 211 | lobby.turns.join({ id: "id-ann", name: "ann", owner: false }); | |
| 212 | lobby.turns.end("id-ann"); | |
| 213 | for (let i = 0; i < 100 && facts.turns()[0]?.releases == null; i++) await Bun.sleep(50); | |
| 214 | await Bun.sleep(200); | |
| 215 | const done = async () => { | |
| 216 | await relay.close(); | |
| 217 | lobby.close(); | |
| 218 | sim.stop(true); | |
| 219 | }; | |
| 220 | return { facts, done }; | |
| 221 | } | |
| 222 | ||
| 223 | test("the relay stores the turn's trace; an arm that 404s stores nothing and the row is still there", async () => { | |
| 224 | const ok = await traceRelay("trace"); | |
| 225 | try { | |
| 226 | expect(ok.facts.traceFor(ok.facts.turns()[0]!.id)).toEqual(trace(4)); | |
| 227 | } finally { | |
| 228 | await ok.done(); | |
| 229 | } | |
| 230 | const none = await traceRelay("404"); | |
| 231 | try { | |
| 232 | expect(none.facts.turns()[0]).toMatchObject({ releases: 1 }); | |
| 233 | expect(none.facts.traceFor(none.facts.turns()[0]!.id)).toBeNull(); | |
| 234 | } finally { | |
| 235 | await none.done(); | |
| 236 | } | |
| 237 | }); |
Software/so101/src/facts.ts +43 −2
| @@ -47,6 +47,32 @@ export function parseCard(body: unknown): TurnCard | null { | ||
| 47 | 47 | return { releases: count(card.releases), snapBacks, jams: snapBacks.jam ?? 0, arriveP95Ms: typeof p95 === "number" && Number.isFinite(p95) ? p95 : null }; |
| 48 | 48 | } |
| 49 | 49 | |
| 50 | /** The newest stored session of the arm's GET /precision that began at or after `sinceMs`: the id its trace is read by. */ | |
| 51 | export function parseSessionId(body: unknown, sinceMs: number): number | null { | |
| 52 | const rows = (body as { sessions?: unknown } | null)?.sessions; | |
| 53 | const last = Array.isArray(rows) ? (rows.at(-1) as { id?: unknown; at?: unknown; seconds?: unknown } | null) : null; | |
| 54 | if (!last || typeof last.id !== "number" || !Number.isSafeInteger(last.id) || last.id < 0) return null; | |
| 55 | const end = typeof last.at === "string" ? Date.parse(last.at) : NaN; | |
| 56 | return end >= sinceMs ? last.id : null; | |
| 57 | } | |
| 58 | ||
| 59 | /** The most a stored trace may take, as JSON; a longer one keeps its first samples. */ | |
| 60 | export const MAX_TRACE_BYTES = 256 * 1024; | |
| 61 | ||
| 62 | /** A trace as the arm served it, cut to fit MAX_TRACE_BYTES; null when it is not one. */ | |
| 63 | export function capTrace(body: unknown): string | null { | |
| 64 | const t = body && typeof body === "object" ? (body as { joints?: unknown }) : null; | |
| 65 | if (!t || !t.joints || typeof t.joints !== "object" || Array.isArray(t.joints)) return null; | |
| 66 | let text = JSON.stringify(body); | |
| 67 | let keep = Math.max(...Object.values(t.joints as Record<string, { intent?: unknown[] }>).map((j) => (Array.isArray(j?.intent) ? j.intent.length : 0)), 0); | |
| 68 | while (text.length > MAX_TRACE_BYTES && keep > 0) { | |
| 69 | keep = Math.floor(keep * Math.min(0.9, MAX_TRACE_BYTES / text.length)); | |
| 70 | const joints = Object.fromEntries(Object.entries(t.joints as Record<string, { intent?: unknown[]; actual?: unknown[] }>).map(([k, j]) => [k, { intent: (j.intent ?? []).slice(0, keep), actual: (j.actual ?? []).slice(0, keep) }])); | |
| 71 | text = JSON.stringify({ ...(body as object), joints, truncated: true }); | |
| 72 | } | |
| 73 | return text.length <= MAX_TRACE_BYTES ? text : null; | |
| 74 | } | |
| 75 | ||
| 50 | 76 | export interface Sample { |
| 51 | 77 | watchers: number; |
| 52 | 78 | drivers: number; |
| @@ -90,7 +116,8 @@ export class Facts { | ||
| 90 | 116 | this..run("CREATE TABLE IF NOT EXISTS events (at INTEGER NOT NULL, commit_id TEXT NOT NULL, kind TEXT NOT NULL, who TEXT)"); |
| 91 | 117 | this..run(`CREATE TABLE IF NOT EXISTS samples (at INTEGER NOT NULL, commit_id TEXT NOT NULL, watchers INTEGER NOT NULL, drivers INTEGER NOT NULL, queue INTEGER NOT NULL, |
| 92 | 118 | link_live INTEGER NOT NULL, fps TEXT NOT NULL)`); |
| 93 | for (const t of ["turns", "events", "samples"]) this..run(`CREATE INDEX IF NOT EXISTS ${t}_at ON ${t} (at)`); | |
| 119 | this..run("CREATE TABLE IF NOT EXISTS traces (turn_id INTEGER PRIMARY KEY, at INTEGER NOT NULL, body TEXT NOT NULL)"); | |
| 120 | for (const t of ["turns", "events", "samples", "traces"]) this..run(`CREATE INDEX IF NOT EXISTS ${t}_at ON ${t} (at)`); | |
| 94 | 121 | this.(() => this.prune()); |
| 95 | 122 | } |
| 96 | 123 | |
| @@ -167,11 +194,25 @@ export class Facts { | ||
| 167 | 194 | this.(() => void this..query("UPDATE turns SET releases = ?, snap_backs = ?, jams = ?, arrive_p95_ms = ? WHERE id = ?").run(card.releases, JSON.stringify(card.snapBacks), card.jams, card.arriveP95Ms, turnId)); |
| 168 | 195 | } |
| 169 | 196 | |
| 197 | /** The arm's trace for a turn, stored with it (capped); a body that is not a trace stores nothing. */ | |
| 198 | trace(turnId: number, body: unknown): void { | |
| 199 | this.(() => { | |
| 200 | const text = capTrace(body); | |
| 201 | if (text) this..query("INSERT OR REPLACE INTO traces (turn_id, at, body) VALUES (?, ?, ?)").run(turnId, this.(), text); | |
| 202 | }); | |
| 203 | } | |
| 204 | ||
| 205 | /** The trace kept for a turn, or null when there is none (the arm had no route, or it is gone at 90 days). */ | |
| 206 | traceFor(turnId: number): unknown | null { | |
| 207 | const r = this.(() => this..query("SELECT body FROM traces WHERE turn_id = ?").get(turnId) as { body: string } | null); | |
| 208 | return r ? JSON.parse(r.body) : null; | |
| 209 | } | |
| 210 | ||
| 170 | 211 | /** Drop what is older than 90 days. */ |
| 171 | 212 | prune(): void { |
| 172 | 213 | const before = this.() - KEEP_MS; |
| 173 | 214 | this. = this.(); |
| 174 | for (const t of ["turns", "events", "samples"]) this..query(`DELETE FROM ${t} WHERE at < ?`).run(before); | |
| 215 | for (const t of ["turns", "events", "samples", "traces"]) this..query(`DELETE FROM ${t} WHERE at < ?`).run(before); | |
| 175 | 216 | } |
| 176 | 217 | |
| 177 | 218 | turns(): TurnRow[] { |
Software/so101/src/precision-history.ts +5 −3
| @@ -83,6 +83,8 @@ export interface Row { | ||
| 83 | 83 | releases: number; |
| 84 | 84 | latencyP95Ms: number | null; |
| 85 | 85 | worstJoint: string | null; |
| 86 | /** A stored session's id, the one /precision/trace takes; the hourly rows have none. */ | |
| 87 | id?: number; | |
| 86 | 88 | } |
| 87 | 89 | |
| 88 | 90 | /** One P experiment as kept: the run's own numbers, tagged with the commit that ran it. */ |
| @@ -103,8 +105,8 @@ export interface Scorecard { | ||
| 103 | 105 | hours: Row[]; |
| 104 | 106 | } |
| 105 | 107 | |
| 106 | type Stored = { at: string; commit: string; sessions: number; seconds: number; tracking_p95_deg: number | null; tracking_max_deg: number | null; arrive_p95_ms: number | null; snap_backs: number; releases: number; latency_p95_ms: number | null; worst_joint: string | null }; | |
| 107 | const fromStored = (r: Stored): Row => ({ at: r.at, commit: r.commit, sessions: r.sessions, seconds: r.seconds, trackingP95Deg: r.tracking_p95_deg, trackingMaxDeg: r.tracking_max_deg, arriveP95Ms: r.arrive_p95_ms, snapBacks: r.snap_backs, releases: r.releases, latencyP95Ms: r.latency_p95_ms, worstJoint: r.worst_joint }); | |
| 108 | type Stored = { id?: number; at: string; commit: string; sessions: number; seconds: number; tracking_p95_deg: number | null; tracking_max_deg: number | null; arrive_p95_ms: number | null; snap_backs: number; releases: number; latency_p95_ms: number | null; worst_joint: string | null }; | |
| 109 | const fromStored = (r: Stored): Row => ({ at: r.at, commit: r.commit, sessions: r.sessions, seconds: r.seconds, trackingP95Deg: r.tracking_p95_deg, trackingMaxDeg: r.tracking_max_deg, arriveP95Ms: r.arrive_p95_ms, snapBacks: r.snap_backs, releases: r.releases, latencyP95Ms: r.latency_p95_ms, worstJoint: r.worst_joint, ...(r.id === undefined ? {} : { id: r.id }) }); | |
| 108 | 110 | |
| 109 | 111 | export class PrecisionHistory { |
| 110 | 112 | readonly commit: string; |
| @@ -199,7 +201,7 @@ export class PrecisionHistory { | ||
| 199 | 201 | rows(table: "sessions" | "hours", limit = 48): Row[] { |
| 200 | 202 | const sql = |
| 201 | 203 | table === "sessions" |
| 202 | ? `SELECT at, commit_id AS "commit", sessions, seconds, tracking_p95_deg, tracking_max_deg, arrive_p95_ms, snap_backs, releases, latency_p95_ms, worst_joint FROM sessions ORDER BY id DESC LIMIT ?` | |
| 204 | ? `SELECT id, at, commit_id AS "commit", sessions, seconds, tracking_p95_deg, tracking_max_deg, arrive_p95_ms, snap_backs, releases, latency_p95_ms, worst_joint FROM sessions ORDER BY id DESC LIMIT ?` | |
| 203 | 205 | : `SELECT at, commit_id AS "commit", sessions, seconds, tracking_p95_deg, tracking_max_deg, arrive_p95_ms, snap_backs, releases, latency_p95_ms, worst_joint FROM hours ORDER BY at DESC LIMIT ?`; |
| 204 | 206 | const rows = this..query(sql).all(limit) as Stored[]; |
| 205 | 207 | return rows.reverse().map(fromStored); |
Software/so101/src/relay.ts +9 −2
| @@ -64,7 +64,7 @@ import { LiveMetrics } from "./live-metrics.ts"; | ||
| 64 | 64 | import { wallMs } from "./clock.ts"; |
| 65 | 65 | import { ArmLink, type LinkTimings } from "./arm-link.ts"; |
| 66 | 66 | import { ArmEdges } from "./arm-edges.ts"; |
| 67 | import { parseCard, type Facts } from "./facts.ts"; | |
| 67 | import { parseCard, parseSessionId, type Facts } from "./facts.ts"; | |
| 68 | 68 | import { MachineFeed } from "./machine-feed.ts"; |
| 69 | 69 | import { refusalBatch } from "./refusals.ts"; |
| 70 | 70 | import { BROWSER_HOP, PAGE_HOP, parseSnapBack } from "./precision.ts"; |
| @@ -1130,8 +1130,15 @@ export async function startRelay(options: RelayOptions): Promise<RunningRelay> { | ||
| 1130 | 1130 | setTimeout(() => { |
| 1131 | 1131 | armPrecision(null) |
| 1132 | 1132 | .then(async (answer) => { |
| 1133 | const card = answer.ok ? parseCard(await answer.json()) : null; | |
| 1133 | const body = answer.ok ? await answer.json() : null; | |
| 1134 | const card = parseCard(body); | |
| 1134 | 1135 | if (card) facts.card(row, card); |
| 1136 | // The turn's trace, once: the arm's newest stored session that began with the turn; an arm without the route stores nothing. | |
| 1137 | const startedAt = Date.parse(facts.turns().find((t) => t.id === row)?.startedAt ?? ""); | |
| 1138 | const session = parseSessionId(body, startedAt); | |
| 1139 | if (session === null) return; | |
| 1140 | const traced = await fetch(`${remote && onArm() ? remote.http : simulation}/precision/trace?session=${session}`, { cache: "no-store", signal: AbortSignal.timeout(2000) }); | |
| 1141 | if (traced.ok) facts.trace(row, await traced.json()); | |
| 1135 | 1142 | }) |
| 1136 | 1143 | .catch(() => undefined); // the row stays without its card |
| 1137 | 1144 | }, 1000).unref?.(); |
new above
offline