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 isPermanentEffectError,28 type MutationDescriptor,29 resolveEffectUnit,30 setMutationRuntime,31} from "../schema/effects.ts";32import { appId, syncing } from "./config.ts";33import {34 recordEffectDebugEvent,35 recordEffectLogEntry,36 recordEffectTrace,37 type RuntimeDiagnostics,38} from "./diagnostics.ts";39import { createDefaultNoticeStorage, type NoticeEntry, NoticeQueue } from "./notice-queue.ts";40import { settleDurableWrite } from "./durability.ts";41import { classifyMutationError } from "./mutation-taxonomy.ts";42import {43 findRuntimeTable,44 getRuntime,45 runtimeRecreatedEvent,46 updateRuntimeDiagnostics,47} from "./runtime.ts";48import { assertSchemaWritable } from "./schema-compat.ts";49import { acquireUpgradeWriteLock } from "./upgrade-coordination.ts";50import { createWriteHandle, type WriteHandle, type WriteHandleController } from "./write-handle.ts";51import {52 createDefaultJournalStorage,53 hashJournalValue,54 type JournalDocument,55 type JournalEffectState,56 journalIdFor,57 type JournalStorage,58 type JournalWriteRecord,59 WriteJournal,60} from "./write-journal.ts";6162/** One write not yet settled, as shown by pending-writes surfaces. */63export type PendingWriteSummary = {64 /** The stable journal id of the write. */65 writeId: string;66 /** The declaring verb's name, or `null` for writes without a verb. */67 verb: string | null;68 /** The written table's name. */69 table: string;70 /** The written row's id. */71 rowId: string;72 /** Which operation the write performed. */73 op: "insert" | "update" | "remove";74 /** Epoch milliseconds when the write was journaled. */75 createdAt: number;76 /**77 * True once the verb's declared lifespan has passed while the write is78 * still pending. Surfacing only: the runtime cannot withdraw a locally79 * accepted write, so an overdue intent is reported, never retired.80 */81 expired: boolean;82};8384/** The reload-safe pending set powering "N changes waiting to sync". */85export type PendingWritesSnapshot = {86 /** How many journaled writes have not settled. */87 count: number;88 /** The pending writes, oldest first. */89 writes: readonly PendingWriteSummary[];90};9192/** Per-row sync state for badges: settled, still waiting, or denied. */93export type RowSyncStatus = "synced" | "waiting" | "rejected";9495/** A write the ledger can perform. */96export type LedgerWriteRequest =97 | { kind: "insert"; table: TableProxy<unknown, unknown>; values: unknown }98 | {99 kind: "update";100 table: TableProxy<unknown, unknown>;101 id: string;102 patch: Record<string, unknown>;103 }104 | { kind: "remove"; table: TableProxy<unknown, unknown>; id: string };105106/** Verb metadata carried by a ledger write. */107export type LedgerWriteOptions = {108 /** The declaring verb's name, or `null` for bare table writes. */109 verb?: string | null;110 /** The effect units journaled with this write. */111 units?: readonly EffectUnit<{ id: string }>[];112 /** The intent's lifespan in milliseconds, or `null` for none. */113 expiresAfterMs?: number | null;114 /** Internal deterministic identity for a chain child. */115 writeId?: string;116 /** Parent obligation retaining this chain child until parent completion. */117 retainedBy?: string | null;118};119120type VendorWrite = {121 batchId?: unknown;122 value?: unknown;123 wait(options: { tier: "local" | "global" }): Promise<unknown>;124};125126type LedgerDb = Pick<Db, "onMutationError"> & {127 insert(table: never, values: never): VendorWrite;128 update(table: never, id: string, patch: never): VendorWrite;129 delete(table: never, id: string): VendorWrite;130 all(query: never, options?: { tier: "local" | "global" }): Promise<unknown[]>;131};132133/** The narrow table surface boot-reconciliation probes query by row id. */134export type ProbeTable = { where(input: Record<string, unknown>): unknown };135136/** Runtime seams used by the ledger; tests inject deterministic values. */137export type WriteLedgerEnvironment = {138 /** Opens (or returns) the vendor database. */139 getDb(): Promise<Db>;140 /** Whether managed sync is configured; read live on every write. */141 syncConfigured(): boolean;142 /** Where the journal persists. */143 storage: JournalStorage;144 /** Resolves a declared table by name for boot-reconciliation probes. */145 resolveTable(name: string): ProbeTable | null;146 /** Resolves a registered effect unit by its durable name. */147 resolveEffectUnit(name: string): EffectUnit<{ id: string }> | null;148 /** Subscribes to runtime replacement, so settlement re-arms on a new client. */149 subscribeRuntimeRecreation(listener: () => void): () => void;150 /** Applies one diagnostics update. */151 updateDiagnostics(update: (diagnostics: RuntimeDiagnostics) => void): void;152 /** Current epoch milliseconds. */153 now(): number;154 /** Delay before settlement retry `attempt`; injectable for deterministic tests. */155 retryDelayMs?: (attempt: number) => number;156 /** How long a global-tier probe may run before it is retried later. */157 probeTimeoutMs?: number;158 /** How often retention is swept while the ledger runs. Default 30 seconds. */159 sweepIntervalMs?: number;160 /**161 * Runs before each write reaches the database: rejects to refuse the write162 * (the schema-compatibility gate — the refusal surfaces through the verb163 * call with no journal entry, stage, or effect), and may resolve with a164 * release function invoked once the write settles locally (the cross-tab165 * write lock). Absent means writes are never guarded.166 */167 guardWrite?(): Promise<(() => void) | void>;168};169170const defaultRetryDelay = (attempt: number): number => Math.min(4000 * (attempt + 1), 30_000);171172function rejectionShape(error: unknown): { code: string; reason: string } | null {173 if (!error || typeof error !== "object") return null;174 const value = error as { name?: unknown; code?: unknown; reason?: unknown };175 if (value.name !== "PersistedWriteRejectedError" || typeof value.code !== "string") return null;176 return { code: value.code, reason: typeof value.reason === "string" ? value.reason : "" };177}178179function normalized(value: unknown): unknown {180 if (value === undefined) return undefined;181 return JSON.parse(JSON.stringify(value));182}183184function hashesMatch(185 hashKey: string,186 journaled: Record<string, string>,187 row: Record<string, unknown>,188): boolean {189 for (const [key, digest] of Object.entries(journaled)) {190 if (key === "id") continue;191 if (hashJournalValue(hashKey, normalized(row[key]) ?? null) !== digest) return false;192 }193 return true;194}195196// The default quarantine bound: failing attempts before an obligation197// retires as failed-permanent instead of re-arming forever.198const defaultMaxAttempts = 5;199200type RetiredEffectStatus = "done" | "expired" | "failed-permanent";201202function isRetired(status: JournalEffectState["status"]): status is RetiredEffectStatus {203 return status === "done" || status === "expired" || status === "failed-permanent";204}205206/**207 * The per-app write ledger. One instance owns the journal, the pending-writes208 * observable, boot reconciliation, and effect delivery. Application code209 * reaches it through verbs and hooks; tests construct isolated instances over210 * an injected {@link WriteLedgerEnvironment}.211 */212export class WriteLedger {213 readonly #environment: WriteLedgerEnvironment;214 readonly #journal: WriteJournal;215 readonly #listeners = new Set<() => void>();216 readonly #running = new Set<string>();217 readonly #probing = new Set<string>();218 readonly #timers = new Set<ReturnType<typeof setTimeout>>();219 readonly #sessionFates = new Map<string, { stage: "synced" | "rejected"; at: number }>();220 // Controllers are the mutator half of issued handles; the ledger is the221 // only holder, which is what keeps handle settlement unforgeable.222 #controllers = new Map<string, WriteHandleController<unknown>>();223 readonly #liveRows = new Map<string, Record<string, unknown>>();224 #sweepTimer: ReturnType<typeof setInterval> | null = null;225 #db: LedgerDb | null = null;226 #stopMutationErrors: (() => void) | null = null;227 #stopRecreation: (() => void) | null = null;228 #armed: Promise<void> | null = null;229 #pendingSnapshot: PendingWritesSnapshot = { count: 0, writes: [] };230 #disposed = false;231232 /** Creates a ledger over one environment; call {@link arm} before relying on boot state. */233 constructor(environment: WriteLedgerEnvironment) {234 this.#environment = environment;235 this.#journal = new WriteJournal(environment.storage);236 }237238 /**239 * Loads the journal and reconciles it against the store: confirmed writes240 * settle as synced, replayed verdicts settle as rejected, unsettled writes241 * stay pending under probes, and settled writes with outstanding effect242 * obligations re-run their handlers. Idempotent.243 */244 arm(): Promise<void> {245 this.#armed ??= this.#reconcile();246 return this.#armed;247 }248249 /** Level-triggered pending-writes state; identity changes only on change. */250 getPendingSnapshot = (): PendingWritesSnapshot => this.#pendingSnapshot;251252 /** Subscribes to ledger changes; the listener runs immediately and on every change. */253 subscribe = (listener: () => void): () => void => {254 this.#listeners.add(listener);255 listener();256 let active = true;257 return () => {258 if (!active) return;259 active = false;260 this.#listeners.delete(listener);261 };262 };263264 /** The sync status of one row, derived from journal entries and session fates. */265 rowStatus(rowId: string): RowSyncStatus {266 for (const entry of Object.values(this.#journal.document.writes)) {267 if (entry.rowId === rowId && entry.stage === "saved") return "waiting";268 }269 for (const entry of Object.values(this.#journal.document.writes)) {270 if (entry.rowId === rowId && entry.stage === "rejected") return "rejected";271 }272 const fate = this.#sessionFates.get(rowId);273 if (fate?.stage === "rejected") return "rejected";274 return "synced";275 }276277 /**278 * Performs one write. A refusal — the write guard's or the runtime's own279 * current-policy check — leaves through the verb call as ordinary author280 * error handling and never creates a journal entry, a stage, or an effect:281 * it throws synchronously when the unguarded database is already open, and282 * fails the returned handle otherwise. The handle resolves at `saved` and283 * settles through `synced` or `rejected`.284 */285 perform<T>(request: LedgerWriteRequest, options: LedgerWriteOptions = {}): WriteHandle<T> {286 if (this.#disposed) throw new Error("write ledger has been disposed");287 const { handle, controller } = createWriteHandle<T>(options.writeId ?? crypto.randomUUID());288 const db = this.#db;289 const guard = this.#environment.guardWrite;290 if (db && !guard) {291 this.#execute(292 db,293 request,294 options,295 handle as WriteHandle<unknown>,296 controller as WriteHandleController<unknown>,297 );298 return handle;299 }300 void (async () => {301 let release: (() => void) | undefined;302 try {303 release = (await guard?.()) ?? undefined;304 } catch (error) {305 // A guard refusal is not a verdict: nothing is journaled and no306 // effect fires; the failure surfaces like any refused mutation.307 this.#environment.updateDiagnostics((diagnostics) => diagnostics.mutationErrors += 1);308 controller.fail(error);309 return;310 }311 try {312 const openDb = await this.#withDb();313 this.#execute(314 openDb,315 request,316 options,317 handle as WriteHandle<unknown>,318 controller as WriteHandleController<unknown>,319 );320 } catch (error) {321 controller.fail(error);322 } finally {323 if (release) {324 // The lock covers the local-durability window: it releases when325 // the write settles locally or fails, matching the quiescence bar326 // an upgrade swap waits on.327 const done = release;328 handle.saved.then(() => done(), () => done());329 }330 }331 })();332 return handle;333 }334335 /** Performs one declared verb call; the verb dispatcher's entry point. */336 performVerb(337 descriptor: MutationDescriptor,338 args: readonly unknown[],339 internal: Pick<LedgerWriteOptions, "writeId" | "retainedBy"> = {},340 ): WriteHandle<unknown> {341 const table = descriptor.op.table as TableProxy<unknown, unknown>;342 const options: LedgerWriteOptions = {343 verb: descriptor.verbName,344 units: descriptor.units,345 expiresAfterMs: descriptor.expiresAfterMs,346 ...internal,347 };348 if (descriptor.op.kind === "insert") {349 return this.perform({ kind: "insert", table, values: args[0] }, options);350 }351 if (descriptor.op.kind === "update") {352 return this.perform({353 kind: "update",354 table,355 id: args[0] as string,356 patch: args[1] as Record<string, unknown>,357 }, options);358 }359 return this.perform({ kind: "remove", table, id: args[0] as string }, options);360 }361362 /** Performs a chain child once and retains its record through parent commit. */363 async performChainedVerb(364 descriptor: MutationDescriptor,365 args: readonly unknown[],366 parentJournalId: string,367 ): Promise<void> {368 const writeId = `chain:${parentJournalId}`;369 if (this.#journal.document.writes[writeId]) {370 await this.#journal.flush();371 return;372 }373 const handle = this.performVerb(descriptor, args, { writeId, retainedBy: parentJournalId });374 await handle.saved;375 await this.#journal.flush();376 }377378 /** Re-attempts outstanding obligations of one effect name after late registration. */379 retryObligationsFor(effectName: string): void {380 for (const entry of Object.values(this.#journal.document.writes)) {381 if (entry.stage === "saved") continue;382 const state = entry.effects[effectName];383 if (state && !isRetired(state.status)) void this.#runObligation(entry, effectName);384 }385 }386387 /** Resolves once scheduled journal persistence has settled; test seam. */388 flush(): Promise<void> {389 return this.#journal.flush();390 }391392 /** Releases listeners and cancels every scheduled probe and retry. */393 dispose(): void {394 if (this.#disposed) return;395 this.#disposed = true;396 if (this.#sweepTimer !== null) clearInterval(this.#sweepTimer);397 this.#sweepTimer = null;398 for (const timer of this.#timers) clearTimeout(timer);399 this.#timers.clear();400 this.#listeners.clear();401 this.#stopMutationErrors?.();402 this.#stopMutationErrors = null;403 this.#stopRecreation?.();404 this.#stopRecreation = null;405 }406407 async #withDb(): Promise<LedgerDb> {408 if (this.#db) return this.#db;409 const db = await this.#environment.getDb() as unknown as LedgerDb;410 if (this.#db !== db) {411 this.#db = db;412 this.#attachMutationErrors(db);413 }414 this.#stopRecreation ??= this.#environment.subscribeRuntimeRecreation(() => {415 this.#db = null;416 this.#stopMutationErrors?.();417 this.#stopMutationErrors = null;418 void this.#withDb().then(() => this.#startProbes(), () => undefined);419 });420 return db;421 }422423 #attachMutationErrors(db: LedgerDb): void {424 // The fallback event fires once per runtime lifetime; the journal records425 // the verdict idempotently on arrival and never waits for a re-fire.426 this.#stopMutationErrors = db.onMutationError((event: MutationErrorEvent) => {427 const batchId = event.batch?.batchId;428 if (batchId === undefined) return;429 const entry = Object.values(this.#journal.document.writes)430 .find((candidate) => candidate.batchId === String(batchId));431 if (entry) this.#applyVerdict(entry, event.code ?? null, event.reason ?? "");432 });433 }434435 async #reconcile(): Promise<void> {436 await this.#journal.load();437 // Retention first: obligations whose delivery window closed while the438 // device was away retire before anything re-arms.439 this.#sweepRetention();440 await this.#withDb();441 if (this.#disposed) return;442 for (const entry of Object.values(this.#journal.document.writes)) {443 if (entry.stage === "saved") continue;444 // A settled write with outstanding obligations means a crash landed445 // between handler start and journal completion: at-least-once delivery446 // re-runs the handlers now.447 this.#fireUnits(entry);448 }449 this.#startProbes();450 if (this.#sweepTimer === null && !this.#disposed) {451 this.#sweepTimer = setInterval(452 () => this.#sweepRetention(),453 this.#environment.sweepIntervalMs ?? 30_000,454 );455 // A periodic bookkeeping timer must never keep a process alive.456 (this.#sweepTimer as unknown as { unref?: () => void }).unref?.();457 }458 }459460 // Retention: delivery windows close, overdue intents surface. Runs at boot461 // and periodically while the ledger lives.462 #sweepRetention(): void {463 if (this.#disposed) return;464 const now = this.#environment.now();465 for (const entry of Object.values(this.#journal.document.writes)) {466 if (entry.stage === "saved") continue;467 for (const [name, state] of Object.entries(entry.effects)) {468 if (isRetired(state.status)) continue;469 if (this.#running.has(journalIdFor(entry.writeId, name))) continue;470 if (state.expiresAt !== null && now > state.expiresAt) {471 this.#retireObligation(entry, name, "expired");472 }473 }474 }475 this.#afterChange();476 }477478 #retireObligation(479 entry: JournalWriteRecord,480 effectName: string,481 status: "expired" | "failed-permanent",482 ): void {483 const state = entry.effects[effectName];484 if (!state || isRetired(state.status)) return;485 this.#journal.update(() => {486 state.status = status;487 });488 this.#environment.updateDiagnostics((diagnostics) => {489 if (status === "expired") diagnostics.expiredObligations += 1;490 else diagnostics.quarantinedObligations += 1;491 });492 }493494 #startProbes(): void {495 for (const entry of Object.values(this.#journal.document.writes)) {496 if (entry.stage === "saved") this.#probe(entry.writeId, 0);497 }498 }499500 #probe(writeId: string, attempt: number): void {501 if (this.#disposed || this.#probing.has(writeId)) return;502 const entry = this.#journal.document.writes[writeId];503 if (!entry || entry.stage !== "saved") return;504 const db = this.#db;505 const table = this.#environment.resolveTable(entry.table);506 if (!db || !table) {507 this.#scheduleProbe(writeId, attempt + 1);508 return;509 }510 this.#probing.add(writeId);511 const timeoutMs = this.#environment.probeTimeoutMs ?? 8000;512 let timer: ReturnType<typeof setTimeout> | undefined;513 const expired = new Promise<null>((resolve) => {514 timer = setTimeout(() => resolve(null), timeoutMs);515 this.#timers.add(timer);516 });517 const read = db.all(table.where({ id: entry.rowId }) as never, { tier: "global" })518 .then((rows) => rows as Array<Record<string, unknown>>, () => null);519 void Promise.race([read, expired]).then((rows) => {520 if (timer !== undefined) {521 clearTimeout(timer);522 this.#timers.delete(timer);523 }524 this.#probing.delete(writeId);525 if (this.#disposed) return;526 const current = this.#journal.document.writes[writeId];527 if (!current || current.stage !== "saved") return;528 if (rows === null) {529 this.#scheduleProbe(writeId, attempt + 1);530 return;531 }532 const row = rows.find((candidate) => candidate.id === current.rowId);533 const confirmed = current.op === "remove" ? row === undefined : row !== undefined &&534 hashesMatch(this.#journal.document.hashKey, current.rowHashes, row);535 if (confirmed) this.#settleSynced(current);536 else this.#scheduleProbe(writeId, attempt + 1);537 });538 }539540 #scheduleProbe(writeId: string, attempt: number): void {541 if (this.#disposed) return;542 const delay = (this.#environment.retryDelayMs ?? defaultRetryDelay)(attempt);543 const timer = setTimeout(() => {544 this.#timers.delete(timer);545 this.#probe(writeId, attempt);546 }, delay);547 this.#timers.add(timer);548 }549550 #execute(551 db: LedgerDb,552 request: LedgerWriteRequest,553 options: LedgerWriteOptions,554 handle: WriteHandle<unknown>,555 controller: WriteHandleController<unknown>,556 ): void {557 let vendor: VendorWrite;558 let rowId: string;559 let rowSnapshot: Record<string, unknown>;560 try {561 if (request.kind === "insert") {562 vendor = db.insert(request.table as never, request.values as never);563 const value = vendor.value as { id: string } | undefined;564 rowId = value?.id ?? "";565 rowSnapshot = { ...(normalized(value) as Record<string, unknown> ?? {}) };566 } else if (request.kind === "update") {567 vendor = db.update(request.table as never, request.id, request.patch as never);568 rowId = request.id;569 rowSnapshot = { id: request.id, ...(normalized(request.patch) as Record<string, unknown>) };570 } else {571 vendor = db.delete(request.table as never, request.id);572 rowId = request.id;573 rowSnapshot = { id: request.id };574 }575 } catch (error) {576 // A current-policy refusal: no batch, no journal entry, no stage. The577 // error leaves through the verb call as ordinary author error handling.578 this.#environment.updateDiagnostics((diagnostics) => diagnostics.mutationErrors += 1);579 throw error;580 }581 // The journal persists keyed hashes, never values: enough for the boot582 // equality probe without a second plaintext copy of row data. The values583 // themselves live only in this session's memory, for handler delivery.584 const hashKey = this.#journal.document.hashKey;585 const rowHashes: Record<string, string> = {};586 for (const [column, value] of Object.entries(rowSnapshot)) {587 if (column === "id") continue;588 rowHashes[column] = hashJournalValue(hashKey, normalized(value) ?? null);589 }590 const createdAt = this.#environment.now();591 const entry: JournalWriteRecord = {592 writeId: handle.writeId,593 retainedBy: options.retainedBy ?? null,594 verb: options.verb ?? null,595 table: (request.table as { _table?: string })._table ?? "",596 op: request.kind,597 rowId,598 batchId: vendor.batchId === undefined ? null : String(vendor.batchId),599 rowHashes,600 stage: "saved",601 cause: null,602 code: null,603 reason: null,604 createdAt,605 expiresAt: options.expiresAfterMs != null ? createdAt + options.expiresAfterMs : null,606 effects: Object.fromEntries(607 (options.units ?? []).map((unit) => [608 unit.effectName,609 {610 status: "pending" as const,611 attempts: 0,612 lastError: null,613 expiresAt: unit.expiresAfterMs != null ? createdAt + unit.expiresAfterMs : null,614 },615 ]),616 ),617 };618 this.#journal.update((document) => {619 document.writes[entry.writeId] = entry;620 });621 this.#liveRows.set(entry.writeId, rowSnapshot);622 this.#controllers.set(entry.writeId, controller);623 controller.setBatchId(entry.batchId);624 // Pre-stage the insert's row value so a fast global confirmation can625 // never resolve the stage promises before the value is known.626 if (request.kind === "insert") controller.advance("saving", vendor.value);627 this.#afterChange();628 void this.#settle(vendor, entry, controller);629 }630631 async #settle(632 vendor: VendorWrite,633 entry: JournalWriteRecord,634 controller: WriteHandleController<unknown>,635 ): Promise<void> {636 const syncConfigured = this.#environment.syncConfigured();637 let value: unknown;638 try {639 value = await settleDurableWrite(640 vendor as { wait(options: { tier: "local" | "global" }): Promise<unknown> },641 (update) => this.#environment.updateDiagnostics(update),642 syncConfigured ? "background" : "none",643 {644 onGlobal: () => this.#settleSynced(entry),645 onGlobalError: (error) => this.#handleSettlementError(vendor, entry, error, 0),646 },647 );648 } catch (error) {649 const rejection = rejectionShape(error);650 if (rejection) {651 // Some drivers surface the adjudicated verdict through the local-tier652 // wait; it is still a verdict, not a local durability failure.653 this.#applyVerdict(entry, rejection.code, rejection.reason);654 return;655 }656 // Local durability failed: the write never became durable, so the657 // journal must not present it as pending sync.658 this.#journal.update((document) => {659 delete document.writes[entry.writeId];660 });661 this.#controllers.delete(entry.writeId);662 controller.fail(error);663 this.#afterChange();664 return;665 }666 controller.advance("saved", value);667 // Without a sync location there is no store to adjudicate: local668 // durability is settlement, and effects run now.669 if (!syncConfigured) this.#settleSynced(entry);670 }671672 #handleSettlementError(673 vendor: VendorWrite,674 entry: JournalWriteRecord,675 error: unknown,676 attempt: number,677 ): void {678 const rejection = rejectionShape(error);679 if (rejection) {680 this.#applyVerdict(entry, rejection.code, rejection.reason);681 if (this.#journal.document.writes[entry.writeId]?.stage === "saved") {682 // A non-permanent verdict keeps the write pending; re-derive later.683 this.#scheduleWaitRetry(vendor, entry, attempt + 1);684 }685 return;686 }687 // A transport failure is not a verdict; the write stays pending.688 this.#scheduleWaitRetry(vendor, entry, attempt + 1);689 }690691 #scheduleWaitRetry(vendor: VendorWrite, entry: JournalWriteRecord, attempt: number): void {692 if (this.#disposed) return;693 const delay = (this.#environment.retryDelayMs ?? defaultRetryDelay)(attempt);694 const timer = setTimeout(() => {695 this.#timers.delete(timer);696 const current = this.#journal.document.writes[entry.writeId];697 if (!current || current.stage !== "saved") return;698 vendor.wait({ tier: "global" }).then(699 () => this.#settleSynced(current),700 (error) => this.#handleSettlementError(vendor, current, error, attempt),701 );702 }, delay);703 this.#timers.add(timer);704 }705706 #settleSynced(entry: JournalWriteRecord): void {707 const current = this.#journal.document.writes[entry.writeId];708 if (!current || current.stage !== "saved") return;709 this.#journal.update(() => {710 current.stage = "synced";711 });712 this.#controllers.get(current.writeId)?.advance("synced");713 this.#fireUnits(current);714 this.#afterChange();715 }716717 #applyVerdict(entry: JournalWriteRecord, code: string | null, reason: string): void {718 const current = this.#journal.document.writes[entry.writeId];719 if (!current || current.stage !== "saved") return;720 if (classifyMutationError(code) !== "permanent") return;721 this.#journal.update(() => {722 current.stage = "rejected";723 current.cause = "denied";724 current.code = code;725 current.reason = reason;726 });727 this.#controllers.get(current.writeId)?.reject({ cause: "denied", code, reason });728 this.#fireUnits(current);729 this.#afterChange();730 }731732 #fireUnits(entry: JournalWriteRecord): void {733 // Units are independent: each runs in its own task with its own journal734 // state, so one handler's failure never blocks another unit, and there735 // are no ordering guarantees between units of one write.736 for (const name of Object.keys(entry.effects)) {737 void this.#runObligation(entry, name);738 }739 }740741 async #runObligation(entry: JournalWriteRecord, effectName: string): Promise<void> {742 const journalId = journalIdFor(entry.writeId, effectName);743 const state = entry.effects[effectName];744 if (!state || isRetired(state.status) || this.#running.has(journalId)) return;745 // A closed delivery window retires the obligation before any handler746 // consideration: the write happened, so compensation would be wrong, and747 // the receiver's idempotency window is assumed gone.748 if (state.expiresAt !== null && this.#environment.now() > state.expiresAt) {749 this.#retireObligation(entry, effectName, "expired");750 this.#afterChange();751 return;752 }753 const unit = this.#environment.resolveEffectUnit(effectName);754 // An unregistered unit stays pending: its declaring module may not have755 // loaded yet, and registration re-attempts outstanding obligations.756 if (!unit) return;757 const fate = entry.stage === "synced" ? "synced" as const : "rejected" as const;758 const handler = fate === "synced" ? unit.handlers.onSynced : unit.handlers.onRejected;759 if (!handler) {760 this.#journal.update((document) => {761 state.status = "done";762 this.#releaseRetainedChildren(document, journalId);763 });764 this.#afterChange();765 return;766 }767 this.#running.add(journalId);768 try {769 const row = await this.#resolveHandlerRow(entry, fate);770 // Attempts are journaled before the handler starts: a crash mid-handler771 // re-runs it at the next boot — the documented at-least-once contract.772 this.#journal.update(() => {773 state.attempts += 1;774 state.status = "pending";775 });776 const context: EffectContext = {777 journalId,778 writeId: entry.writeId,779 verb: entry.verb,780 table: entry.table,781 op: entry.op,782 rowId: entry.rowId,783 writeCreatedAt: entry.createdAt,784 fate,785 cause: fate === "rejected" ? entry.cause ?? "denied" : null,786 code: entry.code,787 reason: entry.reason,788 };789 try {790 await handler(row as EffectRow<{ id: string }>, context);791 this.#journal.update((document) => {792 state.status = "done";793 state.lastError = null;794 this.#releaseRetainedChildren(document, journalId);795 });796 } catch (error) {797 const maxAttempts = unit.maxAttempts ?? defaultMaxAttempts;798 this.#journal.update(() => {799 state.status = "failed";800 state.lastError = error instanceof Error ? error.message : String(error);801 });802 this.#environment.updateDiagnostics((diagnostics) =>803 diagnostics.effectHandlerFailures += 1804 );805 // A handler that declared its failure permanent retires now: retrying806 // a request the receiver will keep refusing only burns the budget and807 // delays the quarantine diagnostic. Otherwise a failing handler808 // re-arms until repeated attempts quarantine it.809 if (isPermanentEffectError(error) || state.attempts >= maxAttempts) {810 this.#retireObligation(entry, effectName, "failed-permanent");811 }812 }813 } finally {814 this.#running.delete(journalId);815 this.#afterChange();816 }817 }818819 /**820 * The row a handler receives. Same-session deliveries pass the in-memory821 * write snapshot. After a reload the journal holds no values: a synced822 * write's row is fetched live from the store (sync confirmed it exists); a823 * rejected write was rolled back by the engine, so compensation receives824 * write identity only — the row id and the structured cause.825 */826 async #resolveHandlerRow(827 entry: JournalWriteRecord,828 fate: "synced" | "rejected",829 ): Promise<Record<string, unknown>> {830 const live = this.#liveRows.get(entry.writeId);831 if (live) return live;832 if (fate === "synced" && entry.op !== "remove") {833 const db = this.#db;834 const table = this.#environment.resolveTable(entry.table);835 if (db && table) {836 try {837 const rows = await db.all(838 table.where({ id: entry.rowId }) as never,839 { tier: "local" },840 ) as Array<Record<string, unknown>>;841 const row = rows.find((candidate) => candidate.id === entry.rowId);842 if (row) return row;843 } catch {844 // Fall through to identity-only delivery below.845 }846 }847 }848 return { id: entry.rowId };849 }850851 #releaseRetainedChildren(document: JournalDocument, parentJournalId: string): void {852 for (const child of Object.values(document.writes)) {853 if (child.retainedBy === parentJournalId) child.retainedBy = null;854 }855 }856857 #afterChange(): void {858 const document = this.#journal.document;859 for (const entry of Object.values(document.writes)) {860 if (entry.stage === "saved") continue;861 const fate = this.#sessionFates.get(entry.rowId);862 if (!fate || fate.at <= entry.createdAt) {863 this.#sessionFates.set(entry.rowId, { stage: entry.stage, at: entry.createdAt });864 }865 const settled = Object.values(entry.effects)866 .every((state) => isRetired(state.status));867 if (settled && !entry.retainedBy) {868 // Fully settled entries leave the journal; failed-but-not-quarantined869 // obligations keep theirs so the next boot can re-arm the handler.870 this.#journal.update(() => {871 delete document.writes[entry.writeId];872 });873 this.#controllers.delete(entry.writeId);874 this.#liveRows.delete(entry.writeId);875 }876 }877 const now = this.#environment.now();878 const pending = Object.values(document.writes)879 .filter((entry) => entry.stage === "saved")880 .sort((left, right) => left.createdAt - right.createdAt)881 .map((entry): PendingWriteSummary => ({882 writeId: entry.writeId,883 verb: entry.verb,884 table: entry.table,885 rowId: entry.rowId,886 op: entry.op,887 createdAt: entry.createdAt,888 expired: entry.expiresAt !== null && now > entry.expiresAt,889 }));890 const changed = pending.length !== this.#pendingSnapshot.count ||891 pending.some((entry, index) =>892 this.#pendingSnapshot.writes[index]?.writeId !== entry.writeId ||893 this.#pendingSnapshot.writes[index]?.expired !== entry.expired894 );895 if (changed) this.#pendingSnapshot = { count: pending.length, writes: pending };896 this.#environment.updateDiagnostics((diagnostics) => {897 diagnostics.journaledPendingWrites = pending.length;898 diagnostics.expiredPendingWrites = pending.filter((entry) => entry.expired).length;899 });900 for (const listener of this.#listeners) listener();901 }902}903904let defaultLedger: WriteLedger | null = null;905906function createDefaultLedger(): WriteLedger {907 const ledger = new WriteLedger({908 getDb: async () => (await getRuntime()).db,909 syncConfigured: syncing,910 storage: createDefaultJournalStorage(appId),911 resolveTable: (name) => findRuntimeTable(name),912 resolveEffectUnit,913 subscribeRuntimeRecreation(listener) {914 globalThis.addEventListener(runtimeRecreatedEvent, listener);915 return () => globalThis.removeEventListener(runtimeRecreatedEvent, listener);916 },917 updateDiagnostics: updateRuntimeDiagnostics,918 now: Date.now,919 // Every write — verbs and table mutations alike — passes the920 // schema-compatibility gate and holds the cross-tab write lock for its921 // local-durability window, exactly like the direct table-store surface.922 async guardWrite() {923 await assertSchemaWritable();924 return await acquireUpgradeWriteLock();925 },926 });927 // Arming is best-effort here: a runtime that cannot open (an unsupported928 // browser, a prerender pass) surfaces through its own boot diagnostics.929 void ledger.arm().catch(() => undefined);930 return ledger;931}932933/** The package-wide ledger for the current document, created and armed lazily. */934export function getWriteLedger(): WriteLedger {935 defaultLedger ??= createDefaultLedger();936 return defaultLedger;937}938939/**940 * Arms the package-wide ledger at boot so journaled writes reconcile and941 * outstanding effect obligations re-run before any island mounts.942 */943export function armWriteLedger(): Promise<void> {944 return getWriteLedger().arm();945}946947let defaultNotices: NoticeQueue | null = null;948let noticeSweepTimer: ReturnType<typeof setInterval> | null = null;949950/**951 * The package-wide durable notice queue behind `s.notice`, created lazily and952 * kept in sync with the `activeNotices` diagnostic. Public read/subscribe/953 * dismiss surfaces ({@link listNotices}, {@link subscribeNotices},954 * {@link dismissNotice}) are re-exported from the runtime entry.955 */956export function getNoticeQueue(): NoticeQueue {957 if (!defaultNotices) {958 const queue = new NoticeQueue(959 createDefaultNoticeStorage(appId),960 Date.now,961 (count) =>962 updateRuntimeDiagnostics((diagnostics) => {963 diagnostics.activeNotices = count;964 }),965 );966 defaultNotices = queue;967 // Load persisted entries; the merged initial count publishes through the968 // queue's own onCountChange. A storage-less or prerender context simply969 // starts empty.970 void queue.load().catch(() => undefined);971 // A TTL-only expiry has no queue mutation to repaint it, so a modest972 // periodic sweep retires expired entries and republishes the count.973 if (typeof setInterval === "function") {974 noticeSweepTimer = setInterval(() => queue.sweep(), 30_000);975 (noticeSweepTimer as { unref?: () => void })?.unref?.();976 }977 }978 return defaultNotices;979}980981/** The live durable notices for the current document. */982export function listNotices(): readonly NoticeEntry[] {983 return getNoticeQueue().list();984}985986/** Subscribes to notice-queue changes; returns an unsubscribe function. */987export function subscribeNotices(listener: () => void): () => void {988 return getNoticeQueue().subscribe(listener);989}990991/** Dismisses one durable notice by id. */992export function dismissNotice(id: string): void {993 getNoticeQueue().dismiss(id);994}995996/** Dismisses every durable notice. */997export function dismissAllNotices(): void {998 getNoticeQueue().dismissAll();999}10001001// The runtime half of the schema-declared verb surface: importing the runtime1002// package installs dispatch, so verbs work wherever islands import hooks.1003setMutationRuntime({1004 dispatch: (descriptor, args) => getWriteLedger().performVerb(descriptor, args),1005 dispatchChained: (descriptor, args, parentJournalId) =>1006 getWriteLedger().performChainedVerb(descriptor, args, parentJournalId),1007 recordLog: (label, context) =>1008 updateRuntimeDiagnostics((diagnostics) =>1009 recordEffectLogEntry(diagnostics, {1010 label,1011 verb: context.verb,1012 table: context.table,1013 rowId: context.rowId,1014 fate: context.fate,1015 at: Date.now(),1016 })1017 ),1018 recordTrace: (label, context) =>1019 updateRuntimeDiagnostics((diagnostics) =>1020 recordEffectTrace(diagnostics, {1021 label,1022 verb: context.verb,1023 rowId: context.rowId,1024 fate: context.fate,1025 durationMs: Math.max(0, Date.now() - context.writeCreatedAt),1026 at: Date.now(),1027 })1028 ),1029 recordDebug: (event, context) =>1030 updateRuntimeDiagnostics((diagnostics) =>1031 recordEffectDebugEvent(diagnostics, {1032 verb: context.verb,1033 journalId: context.journalId,1034 event,1035 at: Date.now(),1036 })1037 ),1038 enqueueNotice: (input, context) =>1039 getNoticeQueue().enqueue({1040 id: context.journalId,1041 message: input.message,1042 tone: input.tone,1043 ttlMs: input.ttlMs,1044 }),1045 applyMark: async (table, rowId, patch) => {1046 try {1047 await getWriteLedger().perform({ kind: "update", table, id: rowId, patch }).saved;1048 } catch {1049 // A mark is best-effort: the status patch may find no row (a row removed1050 // elsewhere) or be denied by policy. Swallow so the mark obligation does1051 // not re-arm and quarantine over a patch that will never land; the1052 // fate-as-data guarantee for must-survive signals belongs to s.notice.1053 }1054 },1055 unitRegistered: (name) => defaultLedger?.retryObligationsFor(name),1056});10571058import.meta.hot?.dispose(() => {1059 defaultLedger?.dispose();1060 defaultLedger = null;1061 defaultNotices = null;1062 if (noticeSweepTimer !== null) {1063 clearInterval(noticeSweepTimer);1064 noticeSweepTimer = null;1065 }1066});