Repository Public
1/**2 * ctx.emit(): persisted events, exactly once across replays.3 *4 * Choice (documented in the contract as "idempotent on replay"): instead of "skip while replaying" —5 * which loses or duplicates the events emitted between the last recorded step and a crash — every6 * emit() call takes the next EMIT INDEX synchronously, in call order (the same determinism rule as step7 * seqs), and `runs.next_emit` records how many have been persisted. An emit whose index is below that8 * watermark is skipped; otherwise the event row and the watermark are written in one fenced9 * transaction. Emits of a run are serialised so the watermark only moves forward by one.10 */11import type { EventKind, Trust } from '@aos/shared';12import { canonicalize } from './canonical.js';13import type { RunCtx } from './context.js';1415export class Emitter {16 private cursor = 0;17 private chain: Promise<void> = Promise.resolve();1819 constructor(20 private readonly ctx: RunCtx,21 private persisted: number,22 ) {}2324 emit(kind: EventKind, payload: Record<string, unknown>, trust: Trust): Promise<void> {25 const idx = this.cursor++;26 const ctx = this.ctx;27 const p = this.chain.then(async () => {28 if (idx < this.persisted) return; // already emitted before a crash / suspension29 ctx.assertOpen();30 await ctx.gate.shared();31 try {32 ctx.assertOpen();33 const body = canonicalize(payload);34 await ctx.fenced(async (tx, run) => {35 if (idx < run.next_emit) return;36 tx.emit({ missionId: ctx.missionId, runId: ctx.runId, kind, trust, payload: body });37 await tx.q('UPDATE runs SET next_emit = $2 WHERE id = $1', [ctx.runId, idx + 1]);38 });39 this.persisted = Math.max(this.persisted, idx + 1);40 } finally {41 ctx.gate.releaseShared();42 }43 });44 this.chain = p.catch(() => undefined);45 return p;46 }47 idle(): Promise<void> {48 return this.chain;49 }50}51