// Watching together in a room (WAI extensions/co-watch.md), in a page. // // A watch room's service serves the source and carries its viewers' events in a log // (co-watch §10). This module is the page's part, in two halves: // // - the rules, with nothing of the page: the policy read (§3.1), controls and presentation // statements written byte for byte (§5.1, §7.1), statements read (§4.2), the controlled // position (§5.4, §6), the checks a control meets before a player applies it (§5.2, §5.3), // the catch-up rule (§9), and the clock estimate (§7.3). Its tests (cowatch.test.mjs) run // them against the corpus wai/live-conformance/cowatch/; // - CoWatch, the player: it enters the room by proving its grant's key over the service's // challenge (live-rooms §10.5.3), follows the log, applies each control at its instant on // its estimate of the room clock, keeps a media element at the controlled position by the // catch-up rule, signs a presentation statement twice a slot under the page's key, and shows // the room's sync: its own error, and each viewer's verdict from the clock's last skew // statement. Taken for gone after a silence, it enters again and signs again, after its new // `join`, the controls the room had not carried. // // Everything here is the platform's: WebCrypto's SHA-256 and Ed25519, an HTMLMediaElement. const subtle = globalThis.crypto.subtle; const enc = new TextEncoder(); export const NORMAL = 1000000; export const ACTION = { play: 1, pause: 2, seek: 3, rate: 4 }; export const ACTION_NAME = { 1: "play", 2: "pause", 3: "seek", 4: "rate" }; export const STATE = { playing: 1, paused: 2, stalled: 3, ended: 4 }; export const VERDICT = { 1: "in_sync", 2: "out_of_sync", 3: "unknown" }; const MAGIC = { policy: "WXP1", control: "WXC1", presentation: "WXS1", skew: "WXK1" }; const PRESENTATION_DOMAIN = enc.encode("wai:co-watch-presentation\x01"); const PROOF_DOMAIN = enc.encode("wai:live-room-proof\x01"); const META = 0x01; export const hex = (b) => Array.from(b, (x) => x.toString(16).padStart(2, "0")).join(""); export const unhex = (h) => new Uint8Array((h.match(/../g) || []).map((x) => parseInt(x, 16))); export function concat(...parts) { const n = parts.reduce((a, p) => a + p.length, 0); const out = new Uint8Array(n); let o = 0; for (const p of parts) { out.set(p, o); o += p.length; } return out; } export async function sha256(...parts) { return new Uint8Array(await subtle.digest("SHA-256", concat(...parts))); } const le = (n, bytes) => { const out = new Uint8Array(bytes); let v = BigInt(n); for (let i = 0; i < bytes; i++) { out[i] = Number(v & 0xffn); v >>= 8n; } return out; }; class Rd { constructor(b, p = 0) { this.b = b; this.p = p; } take(n) { if (this.p + n > this.b.length) throw new Error("short"); const s = this.b.subarray(this.p, this.p + n); this.p += n; return s; } u8() { return this.take(1)[0]; } uint(n) { const s = this.take(n); let v = 0n; for (let i = n - 1; i >= 0; i--) v = (v << 8n) | BigInt(s[i]); return v; } u16() { return Number(this.uint(2)); } u32() { return Number(this.uint(4)); } u64() { return Number(this.uint(8)); } // every instant, position and rate here is below 2^53 done() { return this.p === this.b.length; } } // ---------------------------------------------------------------- §3 the policy /** A co-watch policy's fields (§3.1); its signature is the room's to check. */ export function readPolicy(b) { if (new TextDecoder().decode(b.subarray(0, 4)) !== MAGIC.policy || b[4] !== 1) throw new Error("not a WXP1 policy"); const r = new Rd(b, 5); const p = { room: hex(r.take(32)), prev: hex(r.take(32)), effective: r.u64(), source_type: r.u8(), source: hex(r.take(32)), duration_us: r.u64() }; for (const k of ["lead_ms", "horizon_ms", "min_rate", "max_rate", "period_ms", "window_ms", "settle_ms", "bound_us"]) p[k] = r.u32(); p.quantile = r.u8(); for (const k of ["tolerance_us", "nudge", "seek_us"]) p[k] = r.u32(); r.take(64); if (!r.done()) throw new Error("bytes after the policy"); return p; } /** The policy in effect at `t` (§3.4): the last whose `effective` is at or below it. */ export function policyAt(policies, t) { let k = null; policies.forEach((p, i) => { if (p.effective <= t) k = i; }); return k; } /** An object sequence's digest (§3.5). */ export async function sourceDigest(objects) { return sha256(enc.encode("wai:co-watch-source\x01"), le(objects.length, 4), ...objects.flatMap((o) => [unhex(o.content_hash), le(o.duration_ms, 4)])); } // ---------------------------------------------------------------- §5, §7 statements /** A control (§5.1), 22 bytes. */ export function controlPayload(c) { return concat(enc.encode(MAGIC.control), Uint8Array.of(1, c.action), le(c.at, 8), le(c.arg, 8)); } /** A presentation statement's record from `magic` through `rate` (§7.1). */ export function presentationBody(p) { return concat(enc.encode(MAGIC.presentation), Uint8Array.of(1), unhex(p.room), unhex(p.player), unhex(p.source), le(p.at, 8), le(p.position, 8), Uint8Array.of(p.state), le(p.rate, 4)); } /** The statement, signed with Ed25519 under the player's key (§7.1): 186 bytes. */ export async function signPresentation(p, privateKey) { const body = presentationBody(p); const sig = new Uint8Array(await subtle.sign({ name: "Ed25519" }, privateKey, concat(PRESENTATION_DOMAIN, body))); return concat(body, sig); } /** WebCrypto's key for a 32-byte Ed25519 seed (RFC 8410's PKCS #8 form), and its public key. */ export async function keyFromSeed(seed) { const pkcs8 = concat(Uint8Array.of(0x30, 0x2e, 0x02, 0x01, 0x00, 0x30, 0x05, 0x06, 0x03, 0x2b, 0x65, 0x70, 0x04, 0x22, 0x04, 0x20), seed); const privateKey = await subtle.importKey("pkcs8", pkcs8, { name: "Ed25519" }, true, ["sign"]); const jwk = await subtle.exportKey("jwk", privateKey); const publicKey = Uint8Array.from(atob(jwk.x.replace(/-/g, "+").replace(/_/g, "/")), (c) => c.charCodeAt(0)); return { privateKey, publicKey }; } /** A statement of this extension (§4.1, §4.2) read, or null for another `meta` payload; a * statement that does not read is `{ kind: "malformed" }`. */ export function readStatement(payload) { const magic = new TextDecoder().decode(payload.subarray(0, 4)); if (![MAGIC.control, MAGIC.presentation, MAGIC.skew].includes(magic)) return null; try { if (payload[4] !== 1) throw new Error("version"); const r = new Rd(payload, 5); let s; if (magic === MAGIC.control) { s = { kind: "control", action: r.u8(), at: r.u64(), arg: r.u64() }; const ok = s.action === 1 || s.action === 2 ? s.arg === 0 : s.action === 3 ? true : s.action === 4 ? s.arg >= 1 && s.arg <= 0xffffffff : false; if (!ok) throw new Error("arg"); } else if (magic === MAGIC.presentation) { s = { kind: "presentation", room: hex(r.take(32)), player: hex(r.take(32)), source: hex(r.take(32)), at: r.u64(), position: r.u64(), state: r.u8(), rate: r.u32() }; s.sig = r.take(64); if (!(s.state in { 1: 1, 2: 1, 3: 1, 4: 1 }) || s.rate === 0) throw new Error("state"); } else { s = { kind: "skew", from: r.u64(), to: r.u64(), bound_us: r.u32(), quantile: r.u8(), viewers: [], pairs: [] }; const nv = r.u8(); for (let i = 0; i < nv; i++) { s.viewers.push({ principal: hex(r.take(32)), held: r.u64(), due: r.u16(), samples: r.u16(), counted: r.u16(), p50: r.u32(), pq: r.u32(), max: r.u32(), verdict: VERDICT[r.u8()] }); } const np = r.u8(); for (let i = 0; i < np; i++) s.pairs.push({ a: r.u8(), b: r.u8(), n: r.u16(), p50: r.u32(), pq: r.u32(), max: r.u32(), verdict: VERDICT[r.u8()] }); } if (!r.done()) throw new Error("bytes after"); return s; } catch { return { kind: "malformed" }; } } /** A WEV1 record's fields (live-events §3.1), read as far as this module needs them. */ export function readEvent(b) { const r = new Rd(b, 5); const ev = { room: hex(r.take(32)), principal: hex(r.take(32)), seq: r.u64(), prev: hex(r.take(32)), t: r.u64() }; if (r.u8() === 1) r.take(48); ev.kind = r.u8(); ev.payload = r.take(r.u16()); ev.body = b.subarray(0, r.p); ev.sig = r.take(64); return ev; } // ---------------------------------------------------------------- §5.4, §6 the controlled position function positionOf(st, x, duration) { if (!st.playing) return st.pos; const p = st.pos + Number((BigInt(Math.max(0, x - st.anchor)) * BigInt(st.rate)) / 1000000n); return Math.min(duration, p); } /** The controlled position over the room clock (§6), from the policies and the accepted * controls, each `{ at, t, digest (hex), action, arg }`. */ export class Timeline { constructor(policies, controls) { this.policies = policies; this.starts = []; policies.forEach((p, i) => { if (i === 0 || p.source_type !== policies[i - 1].source_type || p.source !== policies[i - 1].source) this.starts.push(p.effective); }); const key = (c) => [c.at, c.t, c.digest]; this.controls = [...controls].sort((a, b) => { const [x, y] = [key(a), key(b)]; for (let i = 0; i < 3; i++) if (x[i] !== y[i]) return x[i] < y[i] ? -1 : 1; return 0; }); const aspect = { 1: 0, 2: 0, 3: 1, 4: 2 }; this.overridden = new Set(); this.controls.forEach((c, i) => { for (const d of this.controls.slice(i + 1)) { if (d.at !== c.at) break; if (aspect[d.action] === aspect[c.action]) { this.overridden.add(c.digest); break; } } }); const events = [...this.starts.map((s) => ({ x: s, start: true })), ...this.controls.map((c) => ({ x: c.at, c }))]; events.sort((a, b) => a.x - b.x || (a.start ? 0 : 1) - (b.start ? 0 : 1)); this.points = []; let st = null; for (const e of events) { if (e.start) st = { playing: false, rate: NORMAL, anchor: e.x, pos: 0 }; else if (st) { st = { ...st, pos: positionOf(st, e.x, this.durationAt(e.x)), anchor: e.x }; if (e.c.action === 1) st.playing = true; else if (e.c.action === 2) st.playing = false; else if (e.c.action === 3) st.pos = e.c.arg; else st.rate = e.c.arg; } if (st) this.points.push([e.x, st]); } } durationAt(x) { const k = policyAt(this.policies, x); return k == null ? 0 : this.policies[k].duration_us; } stateAt(x) { let st = null; for (const [at, s] of this.points) if (at <= x) st = s; return st; } positionAt(x) { const st = this.stateAt(x); return st ? positionOf(st, x, this.durationAt(x)) : 0; } } /** Why a control is not applied (§5.2, §5.3), or null: `mayControl(action)` says whether its * author may send it, under the policy in effect at its `t`. */ export function controlRefusal(policies, t, c, mayControl) { const k = policyAt(policies, t); if (k == null) return "cw_no_policy"; const p = policies[k]; if (!mayControl(c.action)) return "control_not_permitted"; if (c.at < t + p.lead_ms * 1000) return "control_not_ahead"; if (c.at > t + p.horizon_ms * 1000) return "control_beyond_horizon"; const q = policies[policyAt(policies, c.at)]; if ((c.action === 3 && c.arg > q.duration_us) || (c.action === 4 && (c.arg < q.min_rate || c.arg > q.max_rate))) return "control_out_of_range"; return null; } // ---------------------------------------------------------------- §9 catch-up /** The correction the catch-up rule gives (§9). */ export function catchUp(p, playing, rate, controlled, duration, presented) { const e = presented - controlled; const mag = Math.abs(e); if (!playing || controlled >= duration) return mag > p.tolerance_us ? { action: "seek", value: controlled } : { action: "hold" }; if (mag <= p.tolerance_us) return { action: "rate", value: rate }; if (mag > p.seek_us) return { action: "seek", value: controlled }; const r = e > 0 ? rate - p.nudge : rate + p.nudge; return { action: "rate", value: Math.max(p.min_rate, Math.min(p.max_rate, r)) }; } // ---------------------------------------------------------------- §7.3 the clock /** The offset of the service's clock from this page's, in µs, from round trips each * `{ sent, got, server }` (this page's ms at sending and receiving, the server's µs): the * answer of least round trip, at its midpoint (§7.3). */ export function clockOffset(samples) { const best = samples.reduce((a, s) => (!a || s.got - s.sent < a.got - a.sent ? s : a), null); return Math.round(best.server - ((best.sent + best.got) / 2) * 1000); } // ---------------------------------------------------------------- the player const sleep = (ms) => new Promise((r) => setTimeout(r, ms)); // A slot is stated twice (co-watch §7.3), so timer jitter leaves none empty. const STATES_PER_SLOT = 2; // Statements one post carries at most: a front may hold a request's body to a few KiB. const POST_EVENTS = 4; // Times a page taken for gone enters again with nothing of it carried since its last entry // (taken for gone again at once, as by a second stall) before it stops. const REENTRIES = 3; // The player's wait between two passes of its loop, in ms, and the stalls of the loop a page // keeps (`stalls`). const PASS_MS = 50; const STALLS_KEPT = 256; export class CoWatch { // o: { api (path, body) -> {status, body}, wai, room, t0, seed (Uint8Array), principal (hex), // grant (hex), grantDigest (hex), media (HTMLMediaElement), policies (hex), objects, // sourceUrl, mayControl (principal, action) -> bool, target (the join's proof target, // "//authority/path"), log, onChange } constructor(o) { Object.assign(this, o); this.policyList = o.policies.map((h) => readPolicy(unhex(h))); this.policy = this.policyList[0]; this.offset = 0; this.at = 0; // the next entry of the log this.controls = []; this.timeline = new Timeline(this.policyList, []); this.seq = 0; this.prev = "00".repeat(32); // Where the room's copy of this page's chain ends: the next seq, and the last digest it carried. this.carried = { seq: 0, prev: "00".repeat(32) }; this.lastT = 0; this.joinT = null; this.outbox = []; // each { rec, seq, digest, control }: signed, not yet carried this.reentered = 0; // entries again since the room last carried an event of this page this.stopped = null; this.skews = []; this.others = {}; // principal -> its last statement: at, position, error // `gone`: posts refused `not_present`, the room having taken the page for gone; `rejoins`: // its entries again; `resigned`: controls signed again after one; `dropped`: statements // and controls signed before one and not signed again. `stalls`: the stretches in which // the player's loop did not run (`run`), the last STALLS_KEPT of them. this.stats = { log: 0, controls: 0, applied: 0, refused: {}, statements: 0, posted: 0, post_refused: {}, seeks: 0, nudges: 0, verified_source: false, gone: 0, rejoins: 0, resigned: 0, dropped: 0, followed_at: null, stalls: [], errors: [] }; this.onChange = o.onChange || (() => {}); } /** The room clock now, on this page's estimate (§7.3). */ now() { return this.roomTime(Date.now()); } /** The room clock at `ms` on this page's clock (`Date.now()`), on its estimate. */ roomTime(ms) { return Math.max(0, Math.round(ms * 1000 + this.offset - this.t0)); } async estimateClock() { const samples = []; for (let i = 0; i < 5; i++) { const sent = Date.now(); const r = await this.api("/watch/time"); const got = Date.now(); if (r.status === 200) samples.push({ sent, got, server: r.body.now_us }); } if (samples.length) this.offset = clockOffset(samples); this.stats.offset_us = this.offset; } /** The source, fetched whole and checked by its content hash (§3.5), as the media's own. */ async loadSource() { const r = await fetch(this.sourceUrl, { cache: "no-store" }); const bytes = new Uint8Array(await r.arrayBuffer()); const h = hex(await sha256(bytes)); const obj = this.objects[0]; if (h !== obj.content_hash) throw new Error(`the source's bytes are not the object the policy names: ${h.slice(0, 16)}…`); if (hex(await sourceDigest(this.objects)) !== this.policy.source) throw new Error("the source's objects are not the policy's source"); this.stats.verified_source = true; this.media.src = URL.createObjectURL(new Blob([bytes], { type: r.headers.get("content-type") || "application/octet-stream" })); this.media.preload = "auto"; await new Promise((ok, bad) => { if (this.media.readyState >= 1) return ok(); this.media.addEventListener("loadedmetadata", ok, { once: true }); this.media.addEventListener("error", () => bad(new Error("the source does not play here")), { once: true }); }); } /** Enter: a proof of the grant's key over the service's challenge, for the request's own * method and target (live-rooms §10.5.1, §10.5.3); the clock then signs this page's `join`. */ async join() { const c = await this.api("/watch/challenge"); if (c.status !== 200) throw new Error(`no challenge: ${c.status}`); const grant = unhex(this.grant); const digest = await sha256(grant.subarray(0, grant.length - 64)); const context = enc.encode(`POST ${this.target}`); const key = await keyFromSeed(this.seed); const body = concat(enc.encode("WRP1"), Uint8Array.of(1), digest, unhex(this.principal), unhex(c.body.challenge), le(context.length, 2), context); const sig = new Uint8Array(await subtle.sign({ name: "Ed25519" }, key.privateKey, concat(PROOF_DOMAIN, body))); const r = await this.api("/watch/join", { room: this.room, grant: this.grant, proof: hex(concat(body, sig)) }); if (r.status !== 200) throw new Error(`the room refused the join: ${r.status} ${r.body && r.body.error}`); this.joinT = r.body.t; this.key = key; this.log(`co-watch: joined at room clock ${r.body.t}`); } /** A `meta` event of this page's chain, signed (live-events §3), at `t`: the outbox's entry * for it, with the control it carries, if it is one. */ event(t, payload, control = null) { t = Math.max(t, this.lastT, this.joinT || 0); const seq = this.seq; const rec = this.wai.eventSign(this.seed, unhex(this.room), BigInt(seq), unhex(this.prev), BigInt(t), META, payload); const v = JSON.parse(this.wai.eventRead(rec)); this.seq++; this.prev = v.digest; this.lastT = t; return { rec, seq, digest: v.digest, control }; } /** When a control asked for `at` is signed now: its `t`, and its instant, `at` or, if the * policy's lead (§5.3) no longer allows that, the earliest instant it does. */ controlTime(at) { const t = Math.max(this.now(), this.lastT, this.joinT || 0); return { t, at: Math.max(at, t + this.policy.lead_ms * 1000 + 50000) }; } /** A control signed now, at the instant `when` (`controlTime`) gives. */ signControl(c, when = this.controlTime(c.at)) { const { t, at } = when; return this.event(t, controlPayload({ ...c, at }), { ...c, at }); } /** Ask the room to `action` (ACTION) with `arg`, at the policy's lead from now (§5.3). */ control(action, arg = 0) { const e = this.signControl({ action, at: 0, arg }); this.outbox.push(e); this.flush(); return e.control.at; } /** The outbox's first `n` events carried: the room's copy of the chain ends with them. */ carry(n) { if (!n) return; const last = this.outbox[n - 1]; this.carried = { seq: last.seq + 1, prev: last.digest }; this.outbox.splice(0, n); this.stats.posted += n; } /** After entering again. Every event the room has not carried was signed before its new * `join`, so the room would refuse it as reversed in time: the chain goes on from where the * room's copy ends, at or after the join. The room holds none of those events (it carries * no event it refuses), so signing their positions again leaves it one event at each * (co-watch §10). A control is signed again, at its instant if the policy's lead still * allows it, else at the earliest it does, unless that moves it to or past the instant of * another viewer's control the page holds at or after its own: controls apply in the order * of their instants (§5.4), and the earlier press would then override the later one, so it * is dropped, as a press the room never had. A presentation statement of an instant before * the page was present again is dropped, and the page states anew within half a slot. */ resign() { const pending = this.outbox; this.outbox = []; this.seq = this.carried.seq; this.prev = this.carried.prev; let resigned = 0; for (const e of pending) { if (!e.control) continue; const c = e.control; const when = this.controlTime(c.at); const passed = this.controls.find((x) => x.author !== this.principal && x.at >= c.at && x.at <= when.at); if (passed) { this.log(`co-watch: a ${ACTION_NAME[c.action]} at ${c.at}, signed while the room had taken this page for gone, is dropped: moved to ${when.at}, it would apply after another viewer's ${ACTION_NAME[passed.action]} at ${passed.at}`); continue; } const again = this.signControl(c, when); this.outbox.push(again); if (again.control.at !== c.at) this.log(`co-watch: a ${ACTION_NAME[c.action]} moved from ${c.at} to ${again.control.at}, the earliest the lead allows`); resigned++; } this.stats.rejoins++; this.stats.resigned += resigned; this.stats.dropped += pending.length - resigned; this.log(`co-watch: entered again; ${resigned} control(s) signed again, ${pending.length - resigned} earlier event(s) dropped`); } /** Enter again, the room having taken this page for gone, and sign again what it had not * carried (`resign`). Whether it entered: a join the service could not make now (the clock * could not sign it) leaves the outbox as it is, for the next post to try again. */ async enterAgain() { try { await this.join(); } catch (e) { this.log(`co-watch: entering again failed (${e.message}); the next post tries again`); return false; } this.resign(); return true; } async flush() { if (this.posting || !this.outbox.length) return; this.posting = true; try { while (this.outbox.length) { const batch = this.outbox.slice(0, POST_EVENTS); const r = await this.api("/watch/post", { room: this.room, events: batch.map((e) => hex(e.rec)) }); if (r.status === 200) { this.carry(batch.length); this.reentered = 0; } else if (r.status === 409 && r.body && r.body.error === "not_present" && this.reentered < REENTRIES) { // The room took this page for gone (it was silent too long, as a page on a loaded // machine can be): it enters again, and its chain goes on from what the room carried. // Taken for gone again before anything of it was carried (a second stall, right // after its join), it enters again too, a few times at most. this.carry(r.body.carried || 0); this.stats.gone++; if (await this.enterAgain()) this.reentered++; else await sleep(500); } else if (r.status === 409) { // Those before it were carried; this one never will be, and the chain cannot go // on past it: the page stops stating rather than leave a gap it cannot fill. const why = r.body.error; this.stats.post_refused[why] = (this.stats.post_refused[why] || 0) + 1; this.carry(r.body.carried || 0); this.outbox.shift(); this.stop(`the room refused an event: ${why}`); break; } else { await sleep(500); } } } finally { this.posting = false; } } stop(why) { this.stopped = why; this.stats.errors.push(why); this.log(`co-watch: ${why}`); } /** Follow the room's log, each record checked before anything of it is applied. */ async follow(alive) { this.stats.followed_at = this.now(); while (alive() && !this.stopped) { let r; try { r = await this.api(`/watch?room=${this.room}&from=${this.at}`); } catch { await sleep(1000); continue; } if (r.status === 410) { this.stop("the room was replaced"); break; } if (r.status !== 200 || !r.body) { await sleep(1000); continue; } let changed = false; for (const e of r.body.entries) { if (e.seq !== this.at) { this.stop(`the log skipped from ${this.at} to ${e.seq}`); return; } changed = (await this.onRecord(unhex(e.event))) || changed; this.at++; this.stats.log++; } if (changed) this.timeline = new Timeline(this.policyList, this.controls); this.onChange(); } } refuse(why) { this.stats.refused[why] = (this.stats.refused[why] || 0) + 1; } async onRecord(bytes) { const v = JSON.parse(this.wai.eventRead(bytes)); if (!v.ok) { this.refuse(v.reason); return false; } if (v.room !== this.room) { this.refuse("wrong_room"); return false; } if (v.kind !== META) return false; const s = readStatement(readEvent(bytes).payload); if (!s) return false; if (s.kind === "malformed") { this.refuse("cw_statement_malformed"); return false; } if (s.kind === "control") { this.stats.controls++; const why = controlRefusal(this.policyList, v.t, s, (a) => this.mayControl(v.principal, a)); if (why) { this.refuse(why); return false; } this.controls.push({ at: s.at, t: v.t, digest: v.digest, action: s.action, arg: s.arg, author: v.principal }); return true; } if (s.kind === "skew") { this.skews.push(s); if (this.skews.length > 64) this.skews.shift(); return false; } if (s.player !== v.principal) { this.refuse("presentation_wrong_key"); return false; } this.others[v.principal] = { at: s.at, position: s.position, error: s.position - this.timeline.positionAt(s.at), state: s.state }; return false; } /** Keep the media at the controlled position (§9) and state what it presents (§7.3). * * Each pass of the loop states, if half a slot has passed since its last statement, then * waits PASS_MS. Two statements are so at most half a slot and the gap between two passes * apart, and while no such gap is over half a slot every slot holds one: a slot is left empty * only if no pass began in its second half. The page records each gap between two passes * over half a slot (`stalls`): `{ from, to }`, and how long the wait and the signing in it * took (`timer_us`, `signing_us`), all on its estimate of the room clock. The rest of the gap * is the loop's own work. A wait or a signing that took long is the page's thread held, by * the page's own work or by its machine not running it; the record does not tell which. */ async run(alive) { const per = (this.policy.period_ms * 1000) / STATES_PER_SLOT; let next = 0; let last = null; // the room clock at the previous pass let timer = 0; // how long its wait and its signing took, on the room clock (µs) let signing = 0; while (alive() && !this.stopped) { const x = this.now(); if (last != null && x - last > per) { this.stats.stalls.push({ from: last, to: x, timer_us: timer, signing_us: signing }); if (this.stats.stalls.length > STALLS_KEPT) this.stats.stalls.shift(); } last = x; signing = 0; const st = this.timeline.stateAt(x); const duration = this.timeline.durationAt(x); const controlled = this.timeline.positionAt(x); const presented = Math.round(this.media.currentTime * 1e6); if (st) { if (st.playing && controlled < duration && this.media.paused) this.media.play().catch(() => {}); if ((!st.playing || controlled >= duration) && !this.media.paused) this.media.pause(); const c = catchUp(this.policy, st.playing, st.rate, controlled, duration, presented); if (c.action === "seek") { this.media.currentTime = c.value / 1e6; this.stats.seeks++; } else if (c.action === "rate") { if (c.value !== st.rate) this.stats.nudges++; if (Math.abs(this.media.playbackRate - c.value / 1e6) > 1e-6) this.media.playbackRate = c.value / 1e6; } this.error = presented - controlled; this.controlled = controlled; } if (x >= next && this.key) { next = x + per; const ended = this.media.ended || (st && controlled >= duration); const state = this.media.readyState < 3 && !this.media.paused ? STATE.stalled : ended ? STATE.ended : this.media.paused ? STATE.paused : STATE.playing; const pos = Math.round(this.media.currentTime * 1e6); const body = { room: this.room, player: this.principal, source: this.policy.source, at: x, position: pos, state, rate: Math.max(1, Math.round(this.media.playbackRate * 1e6)) }; const entries = this.stats.rejoins; const s0 = this.now(); const signed = await signPresentation(body, this.key.privateKey); signing = this.now() - s0; if (this.stats.rejoins !== entries) { // The page entered again while it signed (`flush`): `x` is an instant from before its // present join, which it states nothing of. It states anew at once. this.stats.dropped++; next = 0; } else { this.outbox.push(this.event(x, signed)); this.stats.statements++; this.flush(); } } this.onChange(); const w = this.now(); await this.wait(PASS_MS); timer = this.now() - w; } } /** The loop's wait between two passes. */ wait(ms) { return sleep(ms); } /** What a test reads, and what the page shows. */ snapshot() { const last = this.skews.length ? this.skews[this.skews.length - 1] : null; const lastWindow = last ? this.skews.filter((s) => s.from === last.from) : []; const wall = Date.now(); return { principal: this.principal, joined: this.joinT, at: this.at, offset_us: this.offset, room_now: this.roomTime(wall), wall_ms: wall, pass_ms: PASS_MS, controls: this.controls.length, outbox: this.outbox.length, overridden: this.controls.filter((c) => this.timeline.overridden.has(c.digest)).length, controlled: this.controlled ?? null, error_us: this.error ?? null, playing: !this.media.paused, rate: this.media.playbackRate, current_time: this.media.currentTime, skews: this.skews.length, last_window: lastWindow.length ? { from: lastWindow[0].from, to: lastWindow[0].to, bound_us: lastWindow[0].bound_us, viewers: lastWindow.flatMap((s) => s.viewers), pairs: lastWindow.flatMap((s) => s.pairs) } : null, others: this.others, stopped: this.stopped, ...this.stats, }; } }