API index and search · Build metadata

Source snapshot

packages/room/src/authority.ts

1// The room's **write-authority seam** (RINDLE-REALTIME-DESIGN.md §5.3.1): the three2// calls a flushing room makes against the app's authority — claim a placement epoch at3// boot (§2.5), probe the durable lmids before journal replay (§3.3, T2's4// committed-before-crash row), and apply one journaled batch. In every deployment shape5// the counterpart is the API server's `/apply-row-change-txn` host ([`httpAuthority`]);6// the P3 gate drives the same interface against an in-process mock, which is what keeps7// the contract host-independent.8//9// The apply takes the **exact composed body string** the shell journaled — never a10// re-serialized object. One flush id names one immutable byte body forever (§5.311// step 4): a retry, a crash-replay resubmission, and the first send all put identical12// bytes on the wire, which is what makes the authority's `batch_hash` identity check13// (§8.3, T6) meaningful.1415/** One CAS rejection in the authority's `409 conflict` body: the row's authoritative16 *  current image (`null` = absent), keyed by table + pk cells. */17export interface AuthorityConflict {18  table: string;19  pk: unknown[];20  current: unknown[] | null;21}2223/** The apply outcome. Anything else — network failure, a 5xx — is a **throw**: plain24 *  errors retry (same body, same id; dedup absorbs a landed retry), errors marked25 *  `fatal: true` (an identity mismatch — same id, different body — or a malformed26 *  refusal) kill the incarnation loudly instead. */27export type AuthorityApplyResult =28  | { kind: "ok"; applied: boolean; cv?: number }29  | { kind: "conflict"; conflicts: AuthorityConflict[] }30  | { kind: "fenced"; currentEpoch?: number };3132export interface RoomAuthority {33  /** Claim the placement epoch for `doc` (§2.5) — once per shell process, at boot,34   *  AFTER resubmitting the previous incarnation's unconfirmed batches (their recorded35   *  epochs must still be current to land). */36  claimEpoch(doc: string): Promise<number>;37  /** The authority's ledger lmid per client (0 = none) — the boot probe that lets38   *  journal replay absorb already-durable mutations as dedup. */39  lmids(doc: string, clients: string[]): Promise<Record<string, number>>;40  /** Apply one batch: `body` is the exact journaled `/apply-row-change-txn` body41   *  string, sent verbatim. */42  applyRowChangeTxn(body: string): Promise<AuthorityApplyResult>;43}4445/** An apply error the shell must NOT retry (retrying cannot help; something is wrong46 *  with the batch or the room, not the network). */47export interface FatalAuthorityError extends Error {48  fatal: true;49}5051export function fatalAuthorityError(message: string): FatalAuthorityError {52  const e = new Error(message) as FatalAuthorityError;53  e.fatal = true;54  return e;55}5657export interface HttpAuthorityOptions {58  /** The API server's apply endpoint (e.g. `http://…/api/rindle/apply-row-change-txn`). */59  applyUrl: string;60  /** The epoch-claim endpoint (e.g. `http://…/api/rindle/claim-room-epoch`). */61  claimUrl: string;62  /** The lmid-probe endpoint (e.g. `http://…/api/rindle/room-lmids`). */63  lmidsUrl: string;64  /** Extra headers on every call — the epoch-bound flush credential rides here. */65  headers?: Record<string, string>;66  fetch?: typeof fetch;67}6869/** The HTTP [`RoomAuthority`]: the API-server host as the room's sole counterpart70 *  (§5.3.1). Maps `409 {error:"fenced"}` / `409 {error:"conflict"}` to their results,71 *  a `500` mentioning a batch-identity mismatch to a fatal error, and anything else72 *  non-OK to a retryable throw. */73export function httpAuthority(opts: HttpAuthorityOptions): RoomAuthority {74  const doFetch = opts.fetch ?? fetch;75  const headers = { "content-type": "application/json", ...(opts.headers ?? {}) };76  const post = async (url: string, body: string): Promise<Response> =>77    doFetch(url, { method: "POST", headers, body });78  return {79    async claimEpoch(doc) {80      const res = await post(opts.claimUrl, JSON.stringify({ doc }));81      if (!res.ok) {82        throw new Error(`claim-room-epoch failed: ${res.status} ${await res.text()}`);83      }84      const out = (await res.json()) as { epoch?: number };85      if (typeof out.epoch !== "number") {86        throw fatalAuthorityError("claim-room-epoch returned no epoch");87      }88      return out.epoch;89    },90    async lmids(doc, clients) {91      const res = await post(opts.lmidsUrl, JSON.stringify({ doc, clients }));92      if (!res.ok) {93        throw new Error(`room-lmids failed: ${res.status} ${await res.text()}`);94      }95      const out = (await res.json()) as { lmids?: Record<string, number> };96      return out.lmids ?? {};97    },98    async applyRowChangeTxn(body) {99      const res = await post(opts.applyUrl, body);100      if (res.status === 409) {101        const out = (await res.json()) as {102          error?: string;103          currentEpoch?: number;104          conflicts?: AuthorityConflict[];105        };106        if (out.error === "fenced") {107          return { kind: "fenced", currentEpoch: out.currentEpoch };108        }109        if (out.error === "conflict") {110          return { kind: "conflict", conflicts: out.conflicts ?? [] };111        }112        throw new Error(`apply-row-change-txn 409: ${JSON.stringify(out)}`);113      }114      if (!res.ok) {115        const text = await res.text();116        // The §8.3 identity check: same flush id, different body — OUR bug, loud,117        // never retried (a retry re-sends the same mismatched bytes forever).118        if (res.status === 500 && text.includes("batch identity")) {119          throw fatalAuthorityError(`apply-row-change-txn: ${text}`);120        }121        throw new Error(`apply-row-change-txn failed: ${res.status} ${text}`);122      }123      const out = (await res.json()) as { applied?: boolean; cv?: number };124      return { kind: "ok", applied: out.applied === true, cv: out.cv };125    },126  };127}128