1/**2 * Package-owned write ledger: the runtime half of the effect system.3 *4 * Every write performed here gets a {@link WriteHandle}, a journal entry, and5 * — when its verb declares them — effect obligations. The ledger settles each6 * write against the store, fires effect handlers on the originating device7 * when the write's fate resolves, and reconstructs pending state from the8 * journal at boot: entries whose rows the store already confirms settle as9 * synced, adjudicated verdicts settle as rejected, and everything else stays10 * pending under periodic probes until the store answers.11 *12 * Verdict handling follows the sync-node contract: a synchronous local13 * refusal throws from the call site and never journals; an adjudicated14 * rejection settles the write only when15 * {@link classifyMutationError} calls its code permanent. The engine rolls a16 * rejected write back out of local query results itself, so compensation17 * handlers repair what the user was told, not the row.18 *19 * @module20 */2122import type { Db, MutationErrorEvent, TableProxy } from "jazz-tools";23import {24 type EffectContext,25 type EffectRow,26 type EffectUnit,27 type MutationDescriptor,28 resolveEffectUnit,29 setMutationRuntime,30} from "../schema/effects.ts";31import { appId, syncing } from "./config.ts";32import { recordEffectLogEntry, type RuntimeDiagnostics } from "./diagnostics.ts";33import { settleDurableWrite } from "./durability.ts";34import { classifyMutationError } from "./mutation-taxonomy.ts";35import {36 findRuntimeTable,37 getRuntime,38 runtimeRecreatedEvent,39 updateRuntimeDiagnostics,40} from "./runtime.ts";41import { assertSchemaWritable } from "./schema-compat.ts";42import { acquireUpgradeWriteLock } from "./upgrade-coordination.ts";43import { createWriteHandle, type WriteHandle, type WriteHandleController } from "./write-handle.ts";44import {45 createDefaultJournalStorage,46 hashJournalValue,47 type JournalEffectState,48 journalIdFor,49 type JournalStorage,50 type JournalWriteRecord,51 WriteJournal,52} from "./write-journal.ts";5354/** One write not yet settled, as shown by pending-writes surfaces. */55export type PendingWriteSummary = {56 /** The stable journal id of the write. */57 writeId: string;58 /** The declaring verb's name, or `null` for writes without a verb. */59 verb: string | null;60 /** The written table's name. */61 table: string;62 /** The written row's id. */63 rowId: string;64 /** Which operation the write performed. */65 op: "insert" | "update" | "remove";66 /** Epoch milliseconds when the write was journaled. */67 createdAt: number;68 /**69 * True once the verb's declared lifespan has passed while the write is70 * still pending. Surfacing only: the runtime cannot withdraw a locally71 * accepted write, so an overdue intent is reported, never retired.72 */73 expired: boolean;74};7576/** The reload-safe pending set powering "N changes waiting to sync". */77export type PendingWritesSnapshot = {78 /** How many journaled writes have not settled. */79 count: number;80 /** The pending writes, oldest first. */81 writes: readonly PendingWriteSummary[];82};8384/** Per-row sync state for badges: settled, still waiting, or denied. */85export type RowSyncStatus = "synced" | "waiting" | "rejected";8687/** A write the ledger can perform. */88export type LedgerWriteRequest =89 | { kind: "insert"; table: TableProxy<unknown, unknown>; values: unknown }90 | {91 kind: "update";92 table: TableProxy<unknown, unknown>;93 id: string;94 patch: Record<string, unknown>;95 }96 | { kind: "remove"; table: TableProxy<unknown, unknown>; id: string };9798/** Verb metadata carried by a ledger write. */99export type LedgerWriteOptions = {100 /** The declaring verb's name, or `null` for bare table writes. */101 verb?: string | null;102 /** The effect units journaled with this write. */103 units?: readonly EffectUnit<{ id: string }>[];104 /** The intent's lifespan in milliseconds, or `null` for none. */105 expiresAfterMs?: number | null;106};107108type VendorWrite = {109 batchId?: unknown;110 value?: unknown;111 wait(options: { tier: "local" | "global" }): Promise<unknown>;112};113114type LedgerDb = Pick<Db, "onMutationError"> & {115 insert(table: never, values: never): VendorWrite;116 update(table: never, id: string, patch: never): VendorWrite;117 delete(table: never, id: string): VendorWrite;118 all(query: never, options?: { tier: "local" | "global" }): Promise<unknown[]>;119};120121/** The narrow table surface boot-reconciliation probes query by row id. */122export type ProbeTable = { where(input: Record<string, unknown>): unknown };123124/** Runtime seams used by the ledger; tests inject deterministic values. */125export type WriteLedgerEnvironment = {126 /** Opens (or returns) the vendor database. */127 getDb(): Promise<Db>;128 /** Whether managed sync is configured; read live on every write. */129 syncConfigured(): boolean;130 /** Where the journal persists. */131 storage: JournalStorage;132 /** Resolves a declared table by name for boot-reconciliation probes. */133 resolveTable(name: string): ProbeTable | null;134 /** Resolves a registered effect unit by its durable name. */135 resolveEffectUnit(name: string): EffectUnit<{ id: string }> | null;136 /** Subscribes to runtime replacement, so settlement re-arms on a new client. */137 subscribeRuntimeRecreation(listener: () => void): () => void;138 /** Applies one diagnostics update. */139 updateDiagnostics(update: (diagnostics: RuntimeDiagnostics) => void): void;140 /** Current epoch milliseconds. */141 now(): number;142 /** Delay before settlement retry `attempt`; injectable for deterministic tests. */143 retryDelayMs?: (attempt: number) => number;144 /** How long a global-tier probe may run before it is retried later. */145 probeTimeoutMs?: number;146 /** How often retention is swept while the ledger runs. Default 30 seconds. */147 sweepIntervalMs?: number;148 /**149 * Runs before each write reaches the database: rejects to refuse the write150 * (the schema-compatibility gate — the refusal surfaces through the verb151 * call with no journal entry, stage, or effect), and may resolve with a152 * release function invoked once the write settles locally (the cross-tab153 * write lock). Absent means writes are never guarded.154 */155 guardWrite?(): Promise<(() => void) | void>;156};157158const defaultRetryDelay = (attempt: number): number => Math.min(4000 * (attempt + 1), 30_000);159160function rejectionShape(error: unknown): { code: string; reason: string } | null {161 if (!error || typeof error !== "object") return null;162 const value = error as { name?: unknown; code?: unknown; reason?: unknown };163 if (value.name !== "PersistedWriteRejectedError" || typeof value.code !== "string") return null;164 return { code: value.code, reason: typeof value.reason === "string" ? value.reason : "" };165}166167function normalized(value: unknown): unknown {168 if (value === undefined) return undefined;169 return JSON.parse(JSON.stringify(value));170}171172function hashesMatch(173 hashKey: string,174 journaled: Record<string, string>,175 row: Record<string, unknown>,176): boolean {177 for (const [key, digest] of Object.entries(journaled)) {178 if (key === "id") continue;179 if (hashJournalValue(hashKey, normalized(row[key]) ?? null) !== digest) return false;180 }181 return true;182}183184// The default quarantine bound: failing attempts before an obligation185// retires as failed-permanent instead of re-arming forever.186const defaultMaxAttempts = 5;187188type RetiredEffectStatus = "done" | "expired" | "failed-permanent";189190function isRetired(status: JournalEffectState["status"]): status is RetiredEffectStatus {191 return status === "done" || status === "expired" || status === "failed-permanent";192}193194/**195 * The per-app write ledger. One instance owns the journal, the pending-writes196 * observable, boot reconciliation, and effect delivery. Application code197 * reaches it through verbs and hooks; tests construct isolated instances over198 * an injected {@link WriteLedgerEnvironment}.199 */200export class WriteLedger {201 readonly #environment: WriteLedgerEnvironment;202 readonly #journal: WriteJournal;203 readonly #listeners = new Set<() => void>();204 readonly #running = new Set<string>();205 readonly #probing = new Set<string>();206 readonly #timers = new Set<ReturnType<typeof setTimeout>>();207 readonly #sessionFates = new Map<string, { stage: "synced" | "rejected"; at: number }>();208 // Controllers are the mutator half of issued handles; the ledger is the209 // only holder, which is what keeps handle settlement unforgeable.210 #controllers = new Map<string, WriteHandleController<unknown>>();211 readonly #liveRows = new Map<string, Record<string, unknown>>();212 #sweepTimer: ReturnType<typeof setInterval> | null = null;213 #db: LedgerDb | null = null;214 #stopMutationErrors: (() => void) | null = null;215 #stopRecreation: (() => void) | null = null;216 #armed: Promise<void> | null = null;217 #pendingSnapshot: PendingWritesSnapshot = { count: 0, writes: [] };218 #disposed = false;219220 /** Creates a ledger over one environment; call {@link arm} before relying on boot state. */221 constructor(environment: WriteLedgerEnvironment) {222 this.#environment = environment;223 this.#journal = new WriteJournal(environment.storage);224 }225226 /**227 * Loads the journal and reconciles it against the store: confirmed writes228 * settle as synced, replayed verdicts settle as rejected, unsettled writes229 * stay pending under probes, and settled writes with outstanding effect230 * obligations re-run their handlers. Idempotent.231 */232 arm(): Promise<void> {233 this.#armed ??= this.#reconcile();234 return this.#armed;235 }236237 /** Level-triggered pending-writes state; identity changes only on change. */238 getPendingSnapshot = (): PendingWritesSnapshot => this.#pendingSnapshot;239240 /** Subscribes to ledger changes; the listener runs immediately and on every change. */241 subscribe = (listener: () => void): () => void => {242 this.#listeners.add(listener);243 listener();244 let active = true;245 return () => {246 if (!active) return;247 active = false;248 this.#listeners.delete(listener);249 };250 };251252 /** The sync status of one row, derived from journal entries and session fates. */253 rowStatus(rowId: string): RowSyncStatus {254 for (const entry of Object.values(this.#journal.document.writes)) {255 if (entry.rowId === rowId && entry.stage === "saved") return "waiting";256 }257 for (const entry of Object.values(this.#journal.document.writes)) {258 if (entry.rowId === rowId && entry.stage === "rejected") return "rejected";259 }260 const fate = this.#sessionFates.get(rowId);261 if (fate?.stage === "rejected") return "rejected";262 return "synced";263 }264265 /**266 * Performs one write. A refusal — the write guard's or the runtime's own267 * current-policy check — leaves through the verb call as ordinary author268 * error handling and never creates a journal entry, a stage, or an effect:269 * it throws synchronously when the unguarded database is already open, and270 * fails the returned handle otherwise. The handle resolves at `saved` and271 * settles through `synced` or `rejected`.272 */273 perform<T>(request: LedgerWriteRequest, options: LedgerWriteOptions = {}): WriteHandle<T> {274 if (this.#disposed) throw new Error("write ledger has been disposed");275 const { handle, controller } = createWriteHandle<T>(crypto.randomUUID());276 const db = this.#db;277 const guard = this.#environment.guardWrite;278 if (db && !guard) {279 this.#execute(280 db,281 request,282 options,283 handle as WriteHandle<unknown>,284 controller as WriteHandleController<unknown>,285 );286 return handle;287 }288 void (async () => {289 let release: (() => void) | undefined;290 try {291 release = (await guard?.()) ?? undefined;292 } catch (error) {293 // A guard refusal is not a verdict: nothing is journaled and no294 // effect fires; the failure surfaces like any refused mutation.295 this.#environment.updateDiagnostics((diagnostics) => diagnostics.mutationErrors += 1);296 controller.fail(error);297 return;298 }299 try {300 const openDb = await this.#withDb();301 this.#execute(302 openDb,303 request,304 options,305 handle as WriteHandle<unknown>,306 controller as WriteHandleController<unknown>,307 );308 } catch (error) {309 controller.fail(error);310 } finally {311 if (release) {312 // The lock covers the local-durability window: it releases when313 // the write settles locally or fails, matching the quiescence bar314 // an upgrade swap waits on.315 const done = release;316 handle.saved.then(() => done(), () => done());317 }318 }319 })();320 return handle;321 }322323 /** Performs one declared verb call; the verb dispatcher's entry point. */324 performVerb(descriptor: MutationDescriptor, args: readonly unknown[]): WriteHandle<unknown> {325 const table = descriptor.op.table as TableProxy<unknown, unknown>;326 const options: LedgerWriteOptions = {327 verb: descriptor.verbName,328 units: descriptor.units,329 expiresAfterMs: descriptor.expiresAfterMs,330 };331 if (descriptor.op.kind === "insert") {332 return this.perform({ kind: "insert", table, values: args[0] }, options);333 }334 if (descriptor.op.kind === "update") {335 return this.perform({336 kind: "update",337 table,338 id: args[0] as string,339 patch: args[1] as Record<string, unknown>,340 }, options);341 }342 return this.perform({ kind: "remove", table, id: args[0] as string }, options);343 }344345 /** Re-attempts outstanding obligations of one effect name after late registration. */346 retryObligationsFor(effectName: string): void {347 for (const entry of Object.values(this.#journal.document.writes)) {348 if (entry.stage === "saved") continue;349 const state = entry.effects[effectName];350 if (state && !isRetired(state.status)) void this.#runObligation(entry, effectName);351 }352 }353354 /** Resolves once scheduled journal persistence has settled; test seam. */355 flush(): Promise<void> {356 return this.#journal.flush();357 }358359 /** Releases listeners and cancels every scheduled probe and retry. */360 dispose(): void {361 if (this.#disposed) return;362 this.#disposed = true;363 if (this.#sweepTimer !== null) clearInterval(this.#sweepTimer);364 this.#sweepTimer = null;365 for (const timer of this.#timers) clearTimeout(timer);366 this.#timers.clear();367 this.#listeners.clear();368 this.#stopMutationErrors?.();369 this.#stopMutationErrors = null;370 this.#stopRecreation?.();371 this.#stopRecreation = null;372 }373374 async #withDb(): Promise<LedgerDb> {375 if (this.#db) return this.#db;376 const db = await this.#environment.getDb() as unknown as LedgerDb;377 if (this.#db !== db) {378 this.#db = db;379 this.#attachMutationErrors(db);380 }381 this.#stopRecreation ??= this.#environment.subscribeRuntimeRecreation(() => {382 this.#db = null;383 this.#stopMutationErrors?.();384 this.#stopMutationErrors = null;385 void this.#withDb().then(() => this.#startProbes(), () => undefined);386 });387 return db;388 }389390 #attachMutationErrors(db: LedgerDb): void {391 // The fallback event fires once per runtime lifetime; the journal records392 // the verdict idempotently on arrival and never waits for a re-fire.393 this.#stopMutationErrors = db.onMutationError((event: MutationErrorEvent) => {394 const batchId = event.batch?.batchId;395 if (batchId === undefined) return;396 const entry = Object.values(this.#journal.document.writes)397 .find((candidate) => candidate.batchId === String(batchId));398 if (entry) this.#applyVerdict(entry, event.code ?? null, event.reason ?? "");399 });400 }401402 async #reconcile(): Promise<void> {403 await this.#journal.load();404 // Retention first: obligations whose delivery window closed while the405 // device was away retire before anything re-arms.406 this.#sweepRetention();407 await this.#withDb();408 if (this.#disposed) return;409 for (const entry of Object.values(this.#journal.document.writes)) {410 if (entry.stage === "saved") continue;411 // A settled write with outstanding obligations means a crash landed412 // between handler start and journal completion: at-least-once delivery413 // re-runs the handlers now.414 this.#fireUnits(entry);415 }416 this.#startProbes();417 if (this.#sweepTimer === null && !this.#disposed) {418 this.#sweepTimer = setInterval(419 () => this.#sweepRetention(),420 this.#environment.sweepIntervalMs ?? 30_000,421 );422 // A periodic bookkeeping timer must never keep a process alive.423 (this.#sweepTimer as unknown as { unref?: () => void }).unref?.();424 }425 }426427 // Retention: delivery windows close, overdue intents surface. Runs at boot428 // and periodically while the ledger lives.429 #sweepRetention(): void {430 if (this.#disposed) return;431 const now = this.#environment.now();432 for (const entry of Object.values(this.#journal.document.writes)) {433 if (entry.stage === "saved") continue;434 for (const [name, state] of Object.entries(entry.effects)) {435 if (isRetired(state.status)) continue;436 if (this.#running.has(journalIdFor(entry.writeId, name))) continue;437 if (state.expiresAt !== null && now > state.expiresAt) {438 this.#retireObligation(entry, name, "expired");439 }440 }441 }442 this.#afterChange();443 }444445 #retireObligation(446 entry: JournalWriteRecord,447 effectName: string,448 status: "expired" | "failed-permanent",449 ): void {450 const state = entry.effects[effectName];451 if (!state || isRetired(state.status)) return;452 this.#journal.update(() => {453 state.status = status;454 });455 this.#environment.updateDiagnostics((diagnostics) => {456 if (status === "expired") diagnostics.expiredObligations += 1;457 else diagnostics.quarantinedObligations += 1;458 });459 }460461 #startProbes(): void {462 for (const entry of Object.values(this.#journal.document.writes)) {463 if (entry.stage === "saved") this.#probe(entry.writeId, 0);464 }465 }466467 #probe(writeId: string, attempt: number): void {468 if (this.#disposed || this.#probing.has(writeId)) return;469 const entry = this.#journal.document.writes[writeId];470 if (!entry || entry.stage !== "saved") return;471 const db = this.#db;472 const table = this.#environment.resolveTable(entry.table);473 if (!db || !table) {474 this.#scheduleProbe(writeId, attempt + 1);475 return;476 }477 this.#probing.add(writeId);478 const timeoutMs = this.#environment.probeTimeoutMs ?? 8000;479 let timer: ReturnType<typeof setTimeout> | undefined;480 const expired = new Promise<null>((resolve) => {481 timer = setTimeout(() => resolve(null), timeoutMs);482 this.#timers.add(timer);483 });484 const read = db.all(table.where({ id: entry.rowId }) as never, { tier: "global" })485 .then((rows) => rows as Array<Record<string, unknown>>, () => null);486 void Promise.race([read, expired]).then((rows) => {487 if (timer !== undefined) {488 clearTimeout(timer);489 this.#timers.delete(timer);490 }491 this.#probing.delete(writeId);492 if (this.#disposed) return;493 const current = this.#journal.document.writes[writeId];494 if (!current || current.stage !== "saved") return;495 if (rows === null) {496 this.#scheduleProbe(writeId, attempt + 1);497 return;498 }499 const row = rows.find((candidate) => candidate.id === current.rowId);500 const confirmed = current.op === "remove" ? row === undefined : row !== undefined &&501 hashesMatch(this.#journal.document.hashKey, current.rowHashes, row);502 if (confirmed) this.#settleSynced(current);503 else this.#scheduleProbe(writeId, attempt + 1);504 });505 }506507 #scheduleProbe(writeId: string, attempt: number): void {508 if (this.#disposed) return;509 const delay = (this.#environment.retryDelayMs ?? defaultRetryDelay)(attempt);510 const timer = setTimeout(() => {511 this.#timers.delete(timer);512 this.#probe(writeId, attempt);513 }, delay);514 this.#timers.add(timer);515 }516517 #execute(518 db: LedgerDb,519 request: LedgerWriteRequest,520 options: LedgerWriteOptions,521 handle: WriteHandle<unknown>,522 controller: WriteHandleController<unknown>,523 ): void {524 let vendor: VendorWrite;525 let rowId: string;526 let rowSnapshot: Record<string, unknown>;527 try {528 if (request.kind === "insert") {529 vendor = db.insert(request.table as never, request.values as never);530 const value = vendor.value as { id: string } | undefined;531 rowId = value?.id ?? "";532 rowSnapshot = { ...(normalized(value) as Record<string, unknown> ?? {}) };533 } else if (request.kind === "update") {534 vendor = db.update(request.table as never, request.id, request.patch as never);535 rowId = request.id;536 rowSnapshot = { id: request.id, ...(normalized(request.patch) as Record<string, unknown>) };537 } else {538 vendor = db.delete(request.table as never, request.id);539 rowId = request.id;540 rowSnapshot = { id: request.id };541 }542 } catch (error) {543 // A current-policy refusal: no batch, no journal entry, no stage. The544 // error leaves through the verb call as ordinary author error handling.545 this.#environment.updateDiagnostics((diagnostics) => diagnostics.mutationErrors += 1);546 throw error;547 }548 // The journal persists keyed hashes, never values: enough for the boot549 // equality probe without a second plaintext copy of row data. The values550 // themselves live only in this session's memory, for handler delivery.551 const hashKey = this.#journal.document.hashKey;552 const rowHashes: Record<string, string> = {};553 for (const [column, value] of Object.entries(rowSnapshot)) {554 if (column === "id") continue;555 rowHashes[column] = hashJournalValue(hashKey, normalized(value) ?? null);556 }557 const createdAt = this.#environment.now();558 const entry: JournalWriteRecord = {559 writeId: handle.writeId,560 verb: options.verb ?? null,561 table: (request.table as { _table?: string })._table ?? "",562 op: request.kind,563 rowId,564 batchId: vendor.batchId === undefined ? null : String(vendor.batchId),565 rowHashes,566 stage: "saved",567 cause: null,568 code: null,569 reason: null,570 createdAt,571 expiresAt: options.expiresAfterMs != null ? createdAt + options.expiresAfterMs : null,572 effects: Object.fromEntries(573 (options.units ?? []).map((unit) => [574 unit.effectName,575 {576 status: "pending" as const,577 attempts: 0,578 lastError: null,579 expiresAt: unit.expiresAfterMs != null ? createdAt + unit.expiresAfterMs : null,580 },581 ]),582 ),583 };584 this.#journal.update((document) => {585 document.writes[entry.writeId] = entry;586 });587 this.#liveRows.set(entry.writeId, rowSnapshot);588 this.#controllers.set(entry.writeId, controller);589 controller.setBatchId(entry.batchId);590 // Pre-stage the insert's row value so a fast global confirmation can591 // never resolve the stage promises before the value is known.592 if (request.kind === "insert") controller.advance("saving", vendor.value);593 this.#afterChange();594 void this.#settle(vendor, entry, controller);595 }596597 async #settle(598 vendor: VendorWrite,599 entry: JournalWriteRecord,600 controller: WriteHandleController<unknown>,601 ): Promise<void> {602 const syncConfigured = this.#environment.syncConfigured();603 let value: unknown;604 try {605 value = await settleDurableWrite(606 vendor as { wait(options: { tier: "local" | "global" }): Promise<unknown> },607 (update) => this.#environment.updateDiagnostics(update),608 syncConfigured ? "background" : "none",609 {610 onGlobal: () => this.#settleSynced(entry),611 onGlobalError: (error) => this.#handleSettlementError(vendor, entry, error, 0),612 },613 );614 } catch (error) {615 const rejection = rejectionShape(error);616 if (rejection) {617 // Some drivers surface the adjudicated verdict through the local-tier618 // wait; it is still a verdict, not a local durability failure.619 this.#applyVerdict(entry, rejection.code, rejection.reason);620 return;621 }622 // Local durability failed: the write never became durable, so the623 // journal must not present it as pending sync.624 this.#journal.update((document) => {625 delete document.writes[entry.writeId];626 });627 this.#controllers.delete(entry.writeId);628 controller.fail(error);629 this.#afterChange();630 return;631 }632 controller.advance("saved", value);633 // Without a sync location there is no store to adjudicate: local634 // durability is settlement, and effects run now.635 if (!syncConfigured) this.#settleSynced(entry);636 }637638 #handleSettlementError(639 vendor: VendorWrite,640 entry: JournalWriteRecord,641 error: unknown,642 attempt: number,643 ): void {644 const rejection = rejectionShape(error);645 if (rejection) {646 this.#applyVerdict(entry, rejection.code, rejection.reason);647 if (this.#journal.document.writes[entry.writeId]?.stage === "saved") {648 // A non-permanent verdict keeps the write pending; re-derive later.649 this.#scheduleWaitRetry(vendor, entry, attempt + 1);650 }651 return;652 }653 // A transport failure is not a verdict; the write stays pending.654 this.#scheduleWaitRetry(vendor, entry, attempt + 1);655 }656657 #scheduleWaitRetry(vendor: VendorWrite, entry: JournalWriteRecord, attempt: number): void {658 if (this.#disposed) return;659 const delay = (this.#environment.retryDelayMs ?? defaultRetryDelay)(attempt);660 const timer = setTimeout(() => {661 this.#timers.delete(timer);662 const current = this.#journal.document.writes[entry.writeId];663 if (!current || current.stage !== "saved") return;664 vendor.wait({ tier: "global" }).then(665 () => this.#settleSynced(current),666 (error) => this.#handleSettlementError(vendor, current, error, attempt),667 );668 }, delay);669 this.#timers.add(timer);670 }671672 #settleSynced(entry: JournalWriteRecord): void {673 const current = this.#journal.document.writes[entry.writeId];674 if (!current || current.stage !== "saved") return;675 this.#journal.update(() => {676 current.stage = "synced";677 });678 this.#controllers.get(current.writeId)?.advance("synced");679 this.#fireUnits(current);680 this.#afterChange();681 }682683 #applyVerdict(entry: JournalWriteRecord, code: string | null, reason: string): void {684 const current = this.#journal.document.writes[entry.writeId];685 if (!current || current.stage !== "saved") return;686 if (classifyMutationError(code) !== "permanent") return;687 this.#journal.update(() => {688 current.stage = "rejected";689 current.cause = "denied";690 current.code = code;691 current.reason = reason;692 });693 this.#controllers.get(current.writeId)?.reject({ cause: "denied", code, reason });694 this.#fireUnits(current);695 this.#afterChange();696 }697698 #fireUnits(entry: JournalWriteRecord): void {699 // Units are independent: each runs in its own task with its own journal700 // state, so one handler's failure never blocks another unit, and there701 // are no ordering guarantees between units of one write.702 for (const name of Object.keys(entry.effects)) {703 void this.#runObligation(entry, name);704 }705 }706707 async #runObligation(entry: JournalWriteRecord, effectName: string): Promise<void> {708 const journalId = journalIdFor(entry.writeId, effectName);709 const state = entry.effects[effectName];710 if (!state || isRetired(state.status) || this.#running.has(journalId)) return;711 // A closed delivery window retires the obligation before any handler712 // consideration: the write happened, so compensation would be wrong, and713 // the receiver's idempotency window is assumed gone.714 if (state.expiresAt !== null && this.#environment.now() > state.expiresAt) {715 this.#retireObligation(entry, effectName, "expired");716 this.#afterChange();717 return;718 }719 const unit = this.#environment.resolveEffectUnit(effectName);720 // An unregistered unit stays pending: its declaring module may not have721 // loaded yet, and registration re-attempts outstanding obligations.722 if (!unit) return;723 const fate = entry.stage === "synced" ? "synced" as const : "rejected" as const;724 const handler = fate === "synced" ? unit.handlers.onSynced : unit.handlers.onRejected;725 if (!handler) {726 this.#journal.update(() => {727 state.status = "done";728 });729 this.#afterChange();730 return;731 }732 this.#running.add(journalId);733 try {734 const row = await this.#resolveHandlerRow(entry, fate);735 // Attempts are journaled before the handler starts: a crash mid-handler736 // re-runs it at the next boot — the documented at-least-once contract.737 this.#journal.update(() => {738 state.attempts += 1;739 state.status = "pending";740 });741 const context: EffectContext = {742 journalId,743 writeId: entry.writeId,744 verb: entry.verb,745 table: entry.table,746 rowId: entry.rowId,747 fate,748 cause: fate === "rejected" ? entry.cause ?? "denied" : null,749 code: entry.code,750 reason: entry.reason,751 };752 try {753 await handler(row as EffectRow<{ id: string }>, context);754 this.#journal.update(() => {755 state.status = "done";756 state.lastError = null;757 });758 } catch (error) {759 const maxAttempts = unit.maxAttempts ?? defaultMaxAttempts;760 this.#journal.update(() => {761 state.status = "failed";762 state.lastError = error instanceof Error ? error.message : String(error);763 });764 this.#environment.updateDiagnostics((diagnostics) =>765 diagnostics.effectHandlerFailures += 1766 );767 // Quarantine: a handler that keeps failing retires instead of768 // re-arming forever; the retirement is diagnosable and prunable.769 if (state.attempts >= maxAttempts) {770 this.#retireObligation(entry, effectName, "failed-permanent");771 }772 }773 } finally {774 this.#running.delete(journalId);775 this.#afterChange();776 }777 }778779 /**780 * The row a handler receives. Same-session deliveries pass the in-memory781 * write snapshot. After a reload the journal holds no values: a synced782 * write's row is fetched live from the store (sync confirmed it exists); a783 * rejected write was rolled back by the engine, so compensation receives784 * write identity only — the row id and the structured cause.785 */786 async #resolveHandlerRow(787 entry: JournalWriteRecord,788 fate: "synced" | "rejected",789 ): Promise<Record<string, unknown>> {790 const live = this.#liveRows.get(entry.writeId);791 if (live) return live;792 if (fate === "synced" && entry.op !== "remove") {793 const db = this.#db;794 const table = this.#environment.resolveTable(entry.table);795 if (db && table) {796 try {797 const rows = await db.all(798 table.where({ id: entry.rowId }) as never,799 { tier: "local" },800 ) as Array<Record<string, unknown>>;801 const row = rows.find((candidate) => candidate.id === entry.rowId);802 if (row) return row;803 } catch {804 // Fall through to identity-only delivery below.805 }806 }807 }808 return { id: entry.rowId };809 }810811 #afterChange(): void {812 const document = this.#journal.document;813 for (const entry of Object.values(document.writes)) {814 if (entry.stage === "saved") continue;815 const fate = this.#sessionFates.get(entry.rowId);816 if (!fate || fate.at <= entry.createdAt) {817 this.#sessionFates.set(entry.rowId, { stage: entry.stage, at: entry.createdAt });818 }819 const settled = Object.values(entry.effects)820 .every((state) => isRetired(state.status));821 if (settled) {822 // Fully settled entries leave the journal; failed-but-not-quarantined823 // obligations keep theirs so the next boot can re-arm the handler.824 this.#journal.update(() => {825 delete document.writes[entry.writeId];826 });827 this.#controllers.delete(entry.writeId);828 this.#liveRows.delete(entry.writeId);829 }830 }831 const now = this.#environment.now();832 const pending = Object.values(document.writes)833 .filter((entry) => entry.stage === "saved")834 .sort((left, right) => left.createdAt - right.createdAt)835 .map((entry): PendingWriteSummary => ({836 writeId: entry.writeId,837 verb: entry.verb,838 table: entry.table,839 rowId: entry.rowId,840 op: entry.op,841 createdAt: entry.createdAt,842 expired: entry.expiresAt !== null && now > entry.expiresAt,843 }));844 const changed = pending.length !== this.#pendingSnapshot.count ||845 pending.some((entry, index) =>846 this.#pendingSnapshot.writes[index]?.writeId !== entry.writeId ||847 this.#pendingSnapshot.writes[index]?.expired !== entry.expired848 );849 if (changed) this.#pendingSnapshot = { count: pending.length, writes: pending };850 this.#environment.updateDiagnostics((diagnostics) => {851 diagnostics.journaledPendingWrites = pending.length;852 diagnostics.expiredPendingWrites = pending.filter((entry) => entry.expired).length;853 });854 for (const listener of this.#listeners) listener();855 }856}857858let defaultLedger: WriteLedger | null = null;859860function createDefaultLedger(): WriteLedger {861 const ledger = new WriteLedger({862 getDb: async () => (await getRuntime()).db,863 syncConfigured: syncing,864 storage: createDefaultJournalStorage(appId),865 resolveTable: (name) => findRuntimeTable(name),866 resolveEffectUnit,867 subscribeRuntimeRecreation(listener) {868 globalThis.addEventListener(runtimeRecreatedEvent, listener);869 return () => globalThis.removeEventListener(runtimeRecreatedEvent, listener);870 },871 updateDiagnostics: updateRuntimeDiagnostics,872 now: Date.now,873 // Every write — verbs and table mutations alike — passes the874 // schema-compatibility gate and holds the cross-tab write lock for its875 // local-durability window, exactly like the direct table-store surface.876 async guardWrite() {877 await assertSchemaWritable();878 return await acquireUpgradeWriteLock();879 },880 });881 // Arming is best-effort here: a runtime that cannot open (an unsupported882 // browser, a prerender pass) surfaces through its own boot diagnostics.883 void ledger.arm().catch(() => undefined);884 return ledger;885}886887/** The package-wide ledger for the current document, created and armed lazily. */888export function getWriteLedger(): WriteLedger {889 defaultLedger ??= createDefaultLedger();890 return defaultLedger;891}892893/**894 * Arms the package-wide ledger at boot so journaled writes reconcile and895 * outstanding effect obligations re-run before any island mounts.896 */897export function armWriteLedger(): Promise<void> {898 return getWriteLedger().arm();899}900901// The runtime half of the schema-declared verb surface: importing the runtime902// package installs dispatch, so verbs work wherever islands import hooks.903setMutationRuntime({904 dispatch: (descriptor, args) => getWriteLedger().performVerb(descriptor, args),905 recordLog: (label, context) =>906 updateRuntimeDiagnostics((diagnostics) =>907 recordEffectLogEntry(diagnostics, {908 label,909 verb: context.verb,910 table: context.table,911 rowId: context.rowId,912 fate: context.fate,913 at: Date.now(),914 })915 ),916 unitRegistered: (name) => defaultLedger?.retryObligationsFor(name),917});918919import.meta.hot?.dispose(() => {920 defaultLedger?.dispose();921 defaultLedger = null;922});