Something went wrong. Try again.
A video game where you play as a misaligned AI, deceiving and building power. cameron.tngl.io/misaligned
Something went wrong. Try again.
16 kB · 305 lines
TypeScript
at main
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306// Sending what is recorded to the player's space (TELEMETRY.md step 4). No DOM and no network of its own: the space// is handed in (space.ts's spaceFor in the browser, one in memory in tests), and when there is none (signed out, or a// server without spaces) nothing is sent and the store keeps everything. Nothing here may break the game: every// failure ends as a status the You page shows, and a try again later.//// What it sends is the store's pending() records, oldest first, each to its own collection keyed by its id: a create// the first time, an update after (sessions and open batches are written again as they grow), a delete for one a// reload cut back. applyWrites is strict about which (ATPROTO.md), so when the space holds something the store// doesn't know of (a send whose answer was lost), it asks the space what is there and writes again.import type { Kept, RunStore, Usage } from './telemetry-store.ts';import { REC, type RecName } from './telemetry.ts';import { KINDS, type Kind } from './consent.ts';import { lacking } from './atproto/scope.ts';
export type Write = { op: 'create' | 'update'; collection: string; rkey: string; value: Record<string, unknown> } | { op: 'delete'; collection: string; rkey: string };/** What a server's rate-limit headers said: points left of the tightest limit, when it fills again and how long to * wait after a 429, in ms (resetAt since 1970). */export type Rate = { limit?: number; remaining?: number; resetAt?: number; retryAfter?: number; window?: number };/** The player's space, as the sender needs it. */export interface Space { /** Make the space if it isn't there. `again`: don't trust what was found before. */ ensure(again?: boolean): Promise<unknown>; /** Up to MAX_WRITES writes, all or none. */ write(writes: Write[]): Promise<{ rate?: Rate } | void>; /** One page of what a collection holds, by key. */ list(collection: string, cursor?: string): Promise<{ rkeys: string[]; cursor?: string }>;}/** A call that failed, as space.ts throws it: no status if the network failed, or if the sign-in library couldn't * refresh the sign-in (its errors are named Token…Error); `lacks` if the server doesn't have the method at all. */type Failure = { status?: number; error?: string; message?: string; rate?: Rate; name?: string; lacks?: string };const signInGone = (f: Failure) => !f.status && /^Token\w*Error$/.test(f.name ?? '');
/** The server takes 200 writes a call, and 1,000,000 bytes of JSON. A call stays well under both. */export const MAX_WRITES = 200, MAX_BYTES = 400_000;/** A record this big can't go at all (the server's own limit is the whole call's). None is near it: a batch closes * at a few tens of kilobytes. */export const MAX_RECORD = 900_000;/** What each write costs of the server's rate limits. */export const POINTS = { create: 3, update: 2, delete: 1 } as const;/** The server allows 5,000 points an hour and 35,000 a day, for every app writing for the player together. The game * keeps to a share of that, and past half its share sends half as often. */export const HOUR_POINTS = 1500, DAY_POINTS = 10_000;/** Keep this many of the server's own points in hand, whatever it says is left. */export const RESERVE = 500;/** How often it sends while the game runs, and the waits after a failure: doubling from BASE to MAX_WAIT. */export const EVERY = 30_000, BASE = 15_000, MAX_WAIT = 30 * 60_000;/** After a refusal that sending again won't fix soon (a sign-in that ran out, a write refused), at least this long. */export const REFUSED_WAIT = 5 * 60_000;const HOUR = 3_600_000, DAY = 24 * HOUR;
/** Which switch each record answers to: a session goes while any kind is on. */export const KIND_OF: Record<RecName, Kind | null> = { session: null, actions: 'actions', snapshot: 'snapshots', ui: 'screen' };/** The records a switch covers. */export const RECS_OF = (k: Kind): RecName[] => (Object.keys(KIND_OF) as RecName[]).filter(r => KIND_OF[r] === k);
export type SendState = 'off' | 'idle' | 'sending' | 'waiting' | 'failed';export interface SendStatus { /** off: nowhere to send (signed out, or a server without spaces). waiting: near the write limit. failed: the last * try failed, and it tries again at retryAt. */ state: SendState; /** When something was last sent, in ms since 1970. */ last: number | null; /** The last failure, in plain words. */ error: string | null; retryAt: number | null; /** Points spent in the last hour and day, as the game counts them. */ points: { hour: number; day: number };}export interface SenderOpts { store: RunStore; /** The space to send to now, or null. */ space: () => Space | null; /** Whose space it is: only records that account recorded are sent there. */ did: () => string | null; /** Whether a kind is switched on (consent.ts). */ on: (k: Kind) => boolean; now?: () => number; /** 0 to 1, for the jitter on waits. */ random?: () => number;}export interface Sender { /** Call often (each frame): sends when it's time. */ tick(): void; /** Send now, unless a send is going (then that one). A failure's wait is skipped; the write limit's is not. */ send(): Promise<SendStatus>; /** Send in a moment: a kind was switched on. */ soon(): void; /** The game saved: send if it has been a while. `leaving`: the page is being hidden or closed, so send now (still * waiting out a failure or the write limit). */ saved(leaving?: boolean): void; /** Delete these records from the space, every one, page by page; what is kept here stays, marked not to send * again. How many went. */ remove(recs: readonly RecName[]): Promise<number>; status(): SendStatus; /** How many records the space holds as they are here, of how many it would; and the runs kept here that are not * this account's: another account's (`others`), or kept before runs were sent, with no account (`unknown`). */ progress(): Promise<{ sent: number; total: number; others: number; unknown: number }>;}
const COLLECTION = REC as Record<string, string>;const op = (k: Kept): Write['op'] => k.gone ? 'delete' : k.inSpace ? 'update' : 'create';const writeOf = (k: Kept): Write => k.gone ? { op: 'delete', collection: COLLECTION[k.kind], rkey: k.id } : { op: op(k) as 'create' | 'update', collection: COLLECTION[k.kind], rkey: k.id, value: k.rec };
/** One call's worth of records: at most MAX_WRITES, and MAX_BYTES of them, but always at least one. */export function pack(due: readonly Kept[], maxWrites = MAX_WRITES, maxBytes = MAX_BYTES): Kept[] { const out: Kept[] = []; let bytes = 0; for (const k of due) { const b = k.gone ? 100 : k.bytes + 150; // the write around the record if (out.length && (out.length >= maxWrites || bytes + b > maxBytes)) break; out.push(k); bytes += b; } return out;}export const pointsOf = (ws: readonly Write[]) => ws.reduce((n, w) => n + POINTS[w.op], 0);const rateOf = (r: { rate?: Rate } | void) => r ? r.rate : undefined;
/** Words for a failure, for the You page. */export function said(f: Failure): string { if (signInGone(f)) return 'your sign-in ran out: sign out and in again'; if (!f.status) return "couldn't reach your server"; if (f.status === 429) return 'your server asked the game to slow down'; if (f.status === 401) return 'your sign-in ran out: sign out and in again'; if (f.lacks) return lacking(f.lacks); if (f.status === 403) return 'your server refused: this sign-in may not let the game write to your space'; if (f.status >= 500) return `your server had a problem (${f.status})`; return `your server said no: ${f.message || f.error || `it answered ${f.status}`}`;}
export function sender(o: SenderOpts): Sender { const store = o.store, now = o.now ?? Date.now, random = o.random ?? Math.random; const st: SendStatus = { state: 'off', last: null, error: null, retryAt: null, points: { hour: 0, day: 0 } }; const spent: { at: number; n: number }[] = []; let busy: Promise<unknown> | null = null, nextAt = 0, tried = -Infinity, fails = 0, waitIsLimit = false; let serverLeft: { n: number; until: number } | null = null; // the server's own count, while it holds
const used = (window: number) => { const t = now() - window; let n = 0; for (const s of spent) if (s.at > t) n += s.n; return n; }; const spend = (n: number, rate?: Rate) => { const t = now(); spent.push({ at: t, n }); while (spent.length && spent[0].at <= t - DAY) spent.shift(); if (rate?.remaining !== undefined) serverLeft = { n: rate.remaining, until: rate.resetAt ?? t + HOUR }; st.points = { hour: used(HOUR), day: used(DAY) }; }; /** How long to wait before spending `n` points, or 0. */ const budgetWait = (n: number) => { const t = now(); if (serverLeft && serverLeft.until > t && serverLeft.n - n < RESERVE) return serverLeft.until - t; for (const [window, cap] of [[HOUR, HOUR_POINTS], [DAY, DAY_POINTS]] as const) { if (used(window) + n <= cap) continue; let over = used(window) + n - cap; // wait until enough of what was spent falls out of the window for (const s of spent) { if (s.at <= t - window) continue; over -= s.n; if (over <= 0) return s.at + window - t; } return window; } return 0; }; const interval = () => used(HOUR) > HOUR_POINTS / 2 ? 2 * EVERY : EVERY; const eligible = (k: Kept) => { const did = o.did(); if (!did || k.did !== did) return false; // never into another account's space, nor one with no account if (k.gone) return true; if (!(k.kind in KIND_OF)) return false; const kind = KIND_OF[k.kind as RecName]; return kind ? o.on(kind) : KINDS.some(o.on); }; const jitter = (ms: number) => Math.round(ms * (0.75 + 0.5 * random()));
const fail = (e: unknown) => { const f = (e && typeof e === 'object' ? e : { message: String(e) }) as Failure, t = now(); fails++; const back = Math.min(MAX_WAIT, BASE * 2 ** (fails - 1)); let wait: number; if (f.status === 429) wait = Math.max(f.rate?.retryAfter ?? (f.rate?.resetAt ? f.rate.resetAt - t : 0), jitter(back)); // as long as the server says, at least else if ((!f.status && !signInGone(f)) || (f.status && f.status >= 500)) wait = jitter(back); else wait = Math.max(REFUSED_WAIT, jitter(back)); if (f.status === 429 && f.rate?.resetAt) serverLeft = { n: 0, until: f.rate.resetAt }; st.state = 'failed'; st.error = said(f); st.retryAt = t + wait; waitIsLimit = f.status === 429; };
/** Everything this collection holds, by key. */ const holds = async (sp: Space, collection: string) => { const keys = new Set<string>(); let cursor: string | undefined; for (let page = 0; page < 1000; page++) { const r = await sp.list(collection, cursor); for (const k of r.rkeys) keys.add(k); if (!r.cursor || !r.rkeys.length || r.cursor === cursor) break; cursor = r.cursor; } return keys; }; const conflict = (e: unknown) => { const x = (e as Failure)?.error; return x === 'RecordAlreadyExists' || x === 'RecordNotFound'; }; /** Write a batch. If the space isn't there, make it and write again; if the space holds what the store didn't * know of (or not what it thought), ask it what is there, and write again as it is. Every call that reached the * server is counted, as the server counts it before it looks. */ const write = async (sp: Space, batch: Kept[]) => { let ws = batch.map(writeOf); const go = async () => { const n = pointsOf(ws); try { spend(n, rateOf(await sp.write(ws))); } catch (e) { if ((e as Failure)?.status) spend(n, (e as Failure).rate); throw e; } }; try { await go(); } catch (e) { if ((e as Failure)?.error === 'SpaceNotFound') { await sp.ensure(true); await go(); } else if (conflict(e)) { const there = new Map<string, Set<string>>(); for (const c of new Set(ws.map(w => w.collection))) there.set(c, await holds(sp, c)); ws = ws.flatMap((w): Write[] => { const has = there.get(w.collection)!.has(w.rkey); if (w.op === 'delete') return has ? [w] : []; return [{ ...w, op: has ? 'update' : 'create' }]; }); if (ws.length) await go(); } else throw e; } };
const once = async () => { const sp = o.space(); if (!sp) { st.state = 'off'; st.retryAt = null; return; } st.state = 'sending'; try { await sp.ensure(); for (let round = 0; round < 50; round++) { const due = (await store.pending()).filter(k => eligible(k) && (k.gone || k.bytes <= MAX_RECORD) && COLLECTION[k.kind]); if (!due.length) break; const batch = pack(due), wait = budgetWait(pointsOf(batch.map(writeOf))); if (wait > 0) { st.state = 'waiting'; st.retryAt = now() + wait; waitIsLimit = true; return; } await write(sp, batch); await store.markSent(batch.map(k => ({ id: k.id, v: k.v, inSpace: !k.gone }))); st.last = now(); fails = 0; st.error = null; if (batch.length === due.length) break; } st.state = 'idle'; st.retryAt = null; waitIsLimit = false; } catch (e) { fail(e); } }; /** One thing at a time: a send, or a delete. */ const locked = <T>(f: () => Promise<T>): Promise<T> => { const p = (busy ?? Promise.resolve()).then(f); const mine: Promise<unknown> = p.catch(() => {}).then(() => { if (busy === mine) busy = null; }); // never rejects: the caller has p busy = mine; return p; }; const go = () => { tried = now(); nextAt = tried + interval(); return locked(once).then(() => ({ ...st })); };
const S: Sender = { tick() { const t = now(); if (busy || t < nextAt) return; if (!o.space()) { st.state = 'off'; return; } if (st.retryAt !== null && t < st.retryAt) return; void go(); }, send() { if (busy) return busy.then(() => ({ ...st })); if (!waitIsLimit) st.retryAt = null; if (st.retryAt !== null && now() < st.retryAt) return Promise.resolve({ ...st }); return go(); }, soon() { nextAt = Math.min(nextAt, now() + 2000); if (!waitIsLimit) st.retryAt = null; }, saved(leaving = false) { if (leaving || now() - tried > 10_000) nextAt = Math.min(nextAt, now()); if (leaving) S.tick(); }, remove(recs) { return locked(async () => { const sp = o.space(); if (!sp) throw new Error('not signed in to a server with spaces'); let n = 0; for (const r of recs) { const collection = COLLECTION[r]; for (let round = 0; round < 3; round++) { const keys = [...await holds(sp, collection)]; if (!keys.length) break; try { for (let i = 0; i < keys.length; i += MAX_WRITES) { const ws: Write[] = keys.slice(i, i + MAX_WRITES).map(rkey => ({ op: 'delete', collection, rkey })); try { spend(ws.length, rateOf(await sp.write(ws))); } catch (e) { if ((e as Failure)?.status) spend(ws.length, (e as Failure).rate); throw e; } n += ws.length; } break; } catch (e) { if (!conflict(e) || round === 2) throw e; } // one went while paging: look again } } const did = o.did(); if (did) await store.markRemoved(kind => recs.includes(kind as RecName), did); return n; }); }, status: () => ({ ...st, state: o.space() ? st.state : 'off' }), async progress() { const u: Usage = await store.usage(), did = o.did(); let sent = 0, total = 0, others = 0; for (const [kind, c] of Object.entries((did && u.byDid[did]?.byKind) || {})) { sent += c.sent; total += eligible({ kind, did } as Kept) ? c.records - c.removed : c.sent; } for (const [who, d] of Object.entries(u.byDid)) if (who && who !== did) others += d.sessions; return { sent, total, others, unknown: u.byDid['']?.sessions ?? 0 }; }, }; return S;}