API index and search · Build metadata

Source snapshot

packages/room/src/journal.ts

1// The room's pluggable journal (RINDLE-REALTIME-DESIGN.md §2.2, §5.1, §8.1): the2// durable sidecar an **ack** means. The shell appends mutation ENVELOPES under a group3// commit and advances the author's lmid row only after `append` resolves — so `acked`4// survives a room-process crash by construction, replayed by re-invoking the mutators5// against the freshly re-subscribed base (§3.3; recovery is re-invocation, never effect6// replay). Each host brings its own durability: the Durable Object shell backs this7// with per-object transactional storage (P4); the Node test shell defaults to8// `memoryJournal`, whose durability class is exactly "the shell process" — which is9// what makes T2's crash matrix a plain test: hand ONE journal to a second shell and10// the restart-with-journal failure class runs in-process.11//12// With P3's write-behind the journal ALSO holds the built flush batches (§5.3 step 4):13// the exact body bytes are appended BEFORE the first send and dropped once the14// authority confirms, so a crash-replay resubmits the same immutable body — one flush15// id names one byte body forever (the §8.3/T6 identity property). Mutation entries16// still grow without bound here; bounded retention (truncate ≤ the durable watermark)17// is a host-journal concern (P4's storage journal).1819/** One journaled mutation: the wire envelope plus its recorded outcome. Replay applies20 *  the OUTCOME — a non-applied (`rejected`/`deopt`) entry consumes its mid without21 *  running the mutator. */22export interface RoomJournalEntry {23  clientID: string;24  mid: number;25  name: string;26  args: unknown;27  /** The connection's authenticated subject at push time (the lease token's `sub`,28   *  shell-stamped — managed-writes §3.3): recovery is re-invocation, so identity is29   *  an input that must survive the crash. `""` = an entry journaled before the30   *  identity plane existed (mutators see it as unauthenticated). */31  sub: string;32  /** The recorded verdict (H-iv-b): `"applied"` replays by RE-INVOKING the mutator33   *  (§3.3 — and against a moved base the re-invocation may legitimately reject or34   *  DEOPT; the journal record is never rewritten, the shell's recorded-outcome map35   *  reflects what the replaying incarnation produced); `"rejected"` (final — authz /36   *  validation / unknown mutator) and `"deopt"` (the §3.3 commit gate refused; the37   *  client was told to re-route the mutation) both replay as a consumed-mid-no-effect38   *  WITHOUT running anything — re-judging a deopt could invent effects the client39   *  already re-routed elsewhere. Absent = a legacy pre-H-iv-b entry: read through40   *  {@link journalEntryOutcome}. */41  outcome?: "applied" | "rejected" | "deopt";42  /** Legacy pre-H-iv-b flag, superseded by {@link outcome} but still WRITTEN (`true`)43   *  alongside BOTH non-applied outcomes: a legacy reader replays either kind as a44   *  consumed-mid-no-effect, which is exactly right. */45  rejected?: boolean;46}4748/** The ONE reading rule for an entry's verdict across journal generations: `outcome`49 *  when present, else the legacy `rejected` flag, else applied. */50export function journalEntryOutcome(entry: RoomJournalEntry): "applied" | "rejected" | "deopt" {51  return entry.outcome ?? (entry.rejected === true ? "rejected" : "applied");52}5354/** One journaled flush batch: the room's flush-stream position, the placement epoch it55 *  was built under, and the EXACT `/apply-row-change-txn` body string — resubmitted56 *  verbatim, never rebuilt (§5.3 step 4). */57export interface RoomFlushRecord {58  seq: number;59  epoch: number;60  body: string;61}6263export interface RoomJournal {64  /** Append `entries` durably, in order. Resolving is the ack gate (§8.1): the shell65   *  advances lmid rows only after this resolves. A rejection is fatal to the66   *  incarnation — an ack that might not survive must never be sent. */67  append(entries: RoomJournalEntry[]): Promise<void>;68  /** Every entry ever appended, in append order — the boot-time replay source. */69  replay(): Promise<RoomJournalEntry[]>;70  /** Journal one built flush batch, BEFORE its first send. Same durability contract71   *  as `append`: a rejection is fatal (an unjournaled batch must never reach the72   *  wire — a retry could otherwise rebuild different bytes under the same id). */73  appendFlush(record: RoomFlushRecord): Promise<void>;74  /** The authority settled flush `seq` (committed, deduped, or dead) — drop it. */75  confirmFlush(seq: number): Promise<void>;76  /** Unconfirmed flush records in seq order, plus the highest seq ever appended77   *  (0 = none) — the boot-time resubmission source and the seq seed. */78  replayFlushes(): Promise<{ records: RoomFlushRecord[]; maxSeq: number }>;79}8081/** An in-process journal: survives incarnations within one shell process (and, handed82 *  to a second shell, a simulated process crash — the T2 harness). Not durable beyond83 *  the process, by definition. */84export function memoryJournal(): RoomJournal {85  const log: RoomJournalEntry[] = [];86  const flushes = new Map<number, RoomFlushRecord>();87  let maxSeq = 0;88  return {89    append(entries) {90      log.push(...entries);91      return Promise.resolve();92    },93    replay() {94      return Promise.resolve([...log]);95    },96    appendFlush(record) {97      flushes.set(record.seq, record);98      maxSeq = Math.max(maxSeq, record.seq);99      return Promise.resolve();100    },101    confirmFlush(seq) {102      flushes.delete(seq);103      return Promise.resolve();104    },105    replayFlushes() {106      const records = [...flushes.values()].sort((a, b) => a.seq - b.seq);107      return Promise.resolve({ records, maxSeq });108    },109  };110}111