refactor(runtime): one durable-write settlement core for stores and access
- commit
- f32c450c18024d1debac8f0442becd10ec76fd17
- parent
- f0789032eaba46a1d608e6a638fc4e1e2fee5700
- parent
- 0b45dc2e4d8dcdf3b19b50ca0111ac6a6ec8614d
- author
- FelineStateMachine <dami.etoile@gmail.com>
| M | package/ | 33 | +++------- |
| M | package/ | 1 | + |
| A | package/ | 84 | ++++++++++++++++++++++++++ |
| A | package/ | 98 | ++++++++++++++++++++++++++++++ |
| M | package/ | 54 | +++++++--------- |
| M | package/ | 51 | +++++++--------- |
@@ -1,5 +1,6 @@
11 import type { QueryBuilder, TableProxy } from "jazz-tools";22 import { syncing } from "../runtime/config.ts";3+import { type DurableWrite, settleDurableWrite } from "../runtime/durability.ts";34 import { getRuntime, updateRuntimeDiagnostics } from "../runtime/runtime.ts";45 import { AccessError } from "./errors.ts";56 import { decodeSharingIdentity, type SharingIdentity } from "./identity.ts";@@ -27,8 +28,6 @@ export type GroupMembershipRow = Identified & {
2728 can_manage: boolean;2829 };2930 30-type DurableWrite<T> = { wait(options: { tier: "local" | "global" }): Promise<T> };31-3231 function requireSync(operation: string): void {3332 if (!syncing()) {3433 throw new AccessError(@@ -38,33 +37,19 @@ function requireSync(operation: string): void {
3837 }3938 }4039 41-// Share and membership writes settle through the same runtime diagnostics as42-// table mutations, so access operations stay visible to the inspector instead43-// of bypassing pending-write and error counters entirely.40+// Share and membership writes settle through the shared runtime settlement41+// core, so access operations use the same diagnostics accounting as table42+// mutations and stay visible to the inspector. Access writes await the global43+// tier inline: a share is not "done" until the server accepted it.4444 async function settle<T>(write: DurableWrite<T>): Promise<T> {45- updateRuntimeDiagnostics((diagnostics) => diagnostics.pendingLocalWrites += 1);46- let localSettled = false;4745 try {48- await write.wait({ tier: "local" });49- localSettled = true;50- updateRuntimeDiagnostics((diagnostics) => {51- diagnostics.pendingLocalWrites -= 1;52- diagnostics.localWaitCalls += 1;53- diagnostics.pendingGlobalWrites += 1;54- });55- const result = await write.wait({ tier: "global" });56- updateRuntimeDiagnostics((diagnostics) => {57- diagnostics.pendingGlobalWrites -= 1;58- diagnostics.lastWriteDurability = "global";46+ const result = await settleDurableWrite(write, updateRuntimeDiagnostics, "await", {47+ onGlobal: () =>48+ updateRuntimeDiagnostics((diagnostics) => diagnostics.lastWriteDurability = "global"),5949 });6050 return result;6151 } catch (cause) {62- updateRuntimeDiagnostics((diagnostics) => {63- if (localSettled) diagnostics.pendingGlobalWrites -= 1;64- else diagnostics.pendingLocalWrites -= 1;65- diagnostics.mutationErrors += 1;66- diagnostics.lastWriteDurability = "failed";67- });52+ updateRuntimeDiagnostics((diagnostics) => diagnostics.lastWriteDurability = "failed");6853 throw new AccessError(6954 "mutation-rejected",7055 "Jazz rejected the access change. Refresh permissions and membership, then try again.",@@ -29,6 +29,7 @@ export const runtimeFiles = [
2929 "config.ts",3030 "device-capabilities.ts",3131 "diagnostics.ts",32+ "durability.ts",3233 "env.d.ts",3334 "foreground-recovery.ts",3435 "inspector.ts",@@ -0,0 +1,84 @@
1+/**2+ * Package-owned durable-write settlement.3+ *4+ * One implementation of the local→global settlement ladder and its5+ * diagnostics accounting — pending counters, wait counts, and mutation-error6+ * counts — previously triplicated across table stores, mutation stores, and7+ * access operations (where drift left access writes invisible to the8+ * inspector). Snapshot state such as durability labels, error text, and9+ * write-generation guards stays caller-owned through hooks.10+ *11+ * @module12+ */13+14+import type { RuntimeDiagnostics } from "./diagnostics.ts";15+16+/** A vendor write handle that can await local and global durability tiers. */17+export type DurableWrite<T> = {18+ wait(options: { tier: "local" | "global" }): Promise<T>;19+};20+21+/**22+ * How the global tier is handled once local durability settles: not at all23+ * (`none`, sync unconfigured), tracked without blocking the caller24+ * (`background`, UI stores), or awaited inline (`await`, access operations).25+ */26+export type GlobalTierMode = "none" | "background" | "await";27+28+/** Caller hooks projecting settlement outcomes into caller-owned state. */29+export type SettlementHooks = {30+ /** Runs after local durability settles, before any global handling. */31+ onLocal?: () => void;32+ /** Runs when the global tier settles successfully. */33+ onGlobal?: () => void;34+ /** Runs when the global tier rejects; the rejection is already counted. */35+ onGlobalError?: (error: unknown) => void;36+};37+38+/**39+ * Settles one durable write with shared diagnostics accounting. Resolves with40+ * the write's value after local durability (`none`/`background`) or after41+ * global durability (`await`). Rejects on local-tier failure in every mode42+ * and on global-tier failure in `await` mode; a `background` global rejection43+ * reaches {@link SettlementHooks.onGlobalError} only.44+ */45+export async function settleDurableWrite<T>(46+ write: DurableWrite<T>,47+ updateDiagnostics: (update: (diagnostics: RuntimeDiagnostics) => void) => void,48+ globalTier: GlobalTierMode,49+ hooks: SettlementHooks = {},50+): Promise<T> {51+ updateDiagnostics((diagnostics) => diagnostics.pendingLocalWrites += 1);52+ let result: T;53+ try {54+ result = await write.wait({ tier: "local" });55+ } catch (error) {56+ updateDiagnostics((diagnostics) => {57+ diagnostics.pendingLocalWrites -= 1;58+ diagnostics.mutationErrors += 1;59+ });60+ throw error;61+ }62+ updateDiagnostics((diagnostics) => {63+ diagnostics.pendingLocalWrites -= 1;64+ diagnostics.localWaitCalls += 1;65+ });66+ hooks.onLocal?.();67+ if (globalTier === "none") return result;68+ updateDiagnostics((diagnostics) => diagnostics.pendingGlobalWrites += 1);69+ const globalSettled = write.wait({ tier: "global" }).then(70+ () => {71+ hooks.onGlobal?.();72+ },73+ (error) => {74+ updateDiagnostics((diagnostics) => diagnostics.mutationErrors += 1);75+ hooks.onGlobalError?.(error);76+ // Background rejections are fully handled here; only inline waiters77+ // observe the rejection.78+ if (globalTier === "await") throw error;79+ },80+ ).finally(() => updateDiagnostics((diagnostics) => diagnostics.pendingGlobalWrites -= 1));81+ if (globalTier === "await") await globalSettled;82+ else void globalSettled;83+ return result;84+}@@ -0,0 +1,98 @@
1+// Package contract tests for the shared durable-write settlement core.2+import { createDiagnostics, type RuntimeDiagnostics } from "./diagnostics.ts";3+import { settleDurableWrite } from "./durability.ts";4+import { assert, assertCount } from "./test-assert.ts";5+6+const test = (globalThis as unknown as {7+ Deno: { test(name: string, body: () => void | Promise<void>): void };8+}).Deno.test;9+10+function harness(options: { failLocal?: boolean; deferGlobal?: boolean } = {}) {11+ const diagnostics = createDiagnostics();12+ let rejectGlobal!: (error: Error) => void;13+ let resolveGlobal!: () => void;14+ const globalWait = new Promise<void>((resolve, reject) => {15+ resolveGlobal = resolve;16+ rejectGlobal = reject;17+ });18+ const write = {19+ wait: ({ tier }: { tier: "local" | "global" }) => {20+ if (tier === "local") {21+ return options.failLocal22+ ? Promise.reject(new Error("local rejected"))23+ : Promise.resolve("value");24+ }25+ return options.deferGlobal ? globalWait.then(() => "value") : Promise.resolve("value");26+ },27+ };28+ const update = (apply: (diagnostics: RuntimeDiagnostics) => void) => apply(diagnostics);29+ return { diagnostics, write, update, resolveGlobal, rejectGlobal };30+}31+32+async function flush(): Promise<void> {33+ await Promise.resolve();34+ await Promise.resolve();35+}36+37+test("local-only settlement counts one wait and returns the value", async () => {38+ const { diagnostics, write, update } = harness();39+ const value = await settleDurableWrite(write, update, "none");40+ assert(value === "value", "the local value must be returned");41+ assertCount(diagnostics.localWaitCalls, 1, "one local wait must be counted");42+ assertCount(diagnostics.pendingLocalWrites, 0, "no local write may stay pending");43+ assertCount(diagnostics.pendingGlobalWrites, 0, "no global wait may be requested");44+});45+46+test("a local rejection settles the pending counter and counts one error", async () => {47+ const { diagnostics, write, update } = harness({ failLocal: true });48+ let rejected = false;49+ try {50+ await settleDurableWrite(write, update, "background");51+ } catch {52+ rejected = true;53+ }54+ assert(rejected, "a local rejection must reject the settlement");55+ assertCount(diagnostics.pendingLocalWrites, 0, "a failed local write may not stay pending");56+ assertCount(diagnostics.mutationErrors, 1, "the local rejection must be counted once");57+ assertCount(diagnostics.pendingGlobalWrites, 0, "no global tier may start after local failure");58+});59+60+test("background mode resolves at local tier and contains a late global rejection", async () => {61+ const { diagnostics, write, update, rejectGlobal } = harness({ deferGlobal: true });62+ const outcomes: string[] = [];63+ const value = await settleDurableWrite(write, update, "background", {64+ onLocal: () => outcomes.push("local"),65+ onGlobal: () => outcomes.push("global"),66+ onGlobalError: () => outcomes.push("global-error"),67+ });68+ assert(value === "value", "background mode must resolve at local durability");69+ assertCount(diagnostics.pendingGlobalWrites, 1, "the global wait must be tracked");70+ rejectGlobal(new Error("late rejection"));71+ await flush();72+ assert(outcomes.join(",") === "local,global-error", "the rejection must reach its hook");73+ assertCount(diagnostics.mutationErrors, 1, "the global rejection must be counted once");74+ assertCount(diagnostics.pendingGlobalWrites, 0, "the settled global wait must not stay pending");75+});76+77+test("await mode resolves only after global durability and rethrows its rejection", async () => {78+ const success = harness();79+ const outcomes: string[] = [];80+ await settleDurableWrite(success.write, success.update, "await", {81+ onGlobal: () => outcomes.push("global"),82+ });83+ assert(outcomes.join(",") === "global", "await mode must observe global settlement");84+ assertCount(success.diagnostics.pendingGlobalWrites, 0, "the global wait must settle");85+86+ const failure = harness({ deferGlobal: true });87+ const pending = settleDurableWrite(failure.write, failure.update, "await");88+ failure.rejectGlobal(new Error("rejected globally"));89+ let rejected = false;90+ try {91+ await pending;92+ } catch {93+ rejected = true;94+ }95+ assert(rejected, "await mode must reject when the global tier rejects");96+ assertCount(failure.diagnostics.mutationErrors, 1, "the global rejection must be counted");97+ assertCount(failure.diagnostics.pendingGlobalWrites, 0, "the rejected wait must settle");98+});@@ -1,6 +1,7 @@
11 import type { Db, MutationErrorEvent, TableProxy } from "jazz-tools";22 import { syncing } from "./config.ts";33 import type { RuntimeDiagnostics } from "./diagnostics.ts";4+import { settleDurableWrite } from "./durability.ts";45 import { getRuntime, runtimeRecreatedEvent, updateRuntimeDiagnostics } from "./runtime.ts";56 import type { TableRow } from "./table-store.ts";67 @@ -140,46 +141,37 @@ export class TableMutationStore<T extends TableRow, Init> {
140141 async #settle<Result>(mutation: MutationHandle<Result>): Promise<Result> {141142 const generation = ++this.#writeGeneration;142143 this.#trackBatch(mutation);143- this.#updateDiagnostics((diagnostics) => diagnostics.pendingLocalWrites += 1);144144 this.#set({ pending: this.#snapshot.pending + 1, durability: "none", error: null });145145 try {146- const result = await mutation.wait({ tier: "local" });147- this.#updateDiagnostics((diagnostics) => {148- diagnostics.pendingLocalWrites -= 1;149- diagnostics.localWaitCalls += 1;150- });151- this.#set({152- pending: Math.max(0, this.#snapshot.pending - 1),153- ...(generation === this.#writeGeneration ? { durability: "local" as const } : {}),154- });155- if (this.#environment.syncConfigured()) this.#trackGlobal(mutation, generation);156- return result;146+ // Generation guards keep a superseded write's late outcome — success or147+ // rejection — diagnosable without overwriting the newer write's state.148+ return await settleDurableWrite(149+ mutation,150+ (update) => this.#updateDiagnostics(update),151+ this.#environment.syncConfigured() ? "background" : "none",152+ {153+ onLocal: () =>154+ this.#set({155+ pending: Math.max(0, this.#snapshot.pending - 1),156+ ...(generation === this.#writeGeneration ? { durability: "local" as const } : {}),157+ }),158+ onGlobal: () => {159+ if (generation === this.#writeGeneration) {160+ this.#set({ durability: "global", error: null });161+ }162+ },163+ onGlobalError: (error) => {164+ if (generation === this.#writeGeneration) this.#fail(error);165+ },166+ },167+ );157168 } catch (error) {158- this.#updateDiagnostics((diagnostics) => {159- diagnostics.pendingLocalWrites -= 1;160- diagnostics.mutationErrors += 1;161- });162169 this.#set({ pending: Math.max(0, this.#snapshot.pending - 1) });163170 this.#fail(error);164171 throw error;165172 }166173 }167174 168- #trackGlobal<Result>(mutation: MutationHandle<Result>, generation: number): void {169- this.#updateDiagnostics((diagnostics) => diagnostics.pendingGlobalWrites += 1);170- void mutation.wait({ tier: "global" }).then(171- () => {172- if (generation === this.#writeGeneration) this.#set({ durability: "global", error: null });173- },174- (error) => {175- this.#updateDiagnostics((diagnostics) => diagnostics.mutationErrors += 1);176- // Match the success guard: a superseded write's late rejection is177- // diagnosable but must not overwrite the newer write's durability.178- if (generation === this.#writeGeneration) this.#fail(error);179- },180- ).finally(() => this.#updateDiagnostics((diagnostics) => diagnostics.pendingGlobalWrites -= 1));181- }182-183175 #start(): void {184176 this.#stopRuntimeRecreation = this.#environment.subscribeRuntimeRecreation(() =>185177 this.#attach()@@ -1,6 +1,7 @@
11 import type { Db, MutationErrorEvent, QueryBuilder, TableProxy } from "jazz-tools";22 // Package-owned reactive table store.33 import type { RuntimeDiagnostics } from "./diagnostics.ts";4+import { settleDurableWrite } from "./durability.ts";45 56 /** The minimum shape every persisted row exposes to the framework. */67 export type TableRow = { id: string };@@ -173,12 +174,14 @@ export class TableStore<T extends TableRow, Init> {
173174 }174175 }175176 177+ #updateDiagnostics = (update: (diagnostics: RuntimeDiagnostics) => void): void => {178+ update(this.#diagnostics);179+ this.#diagnosticsChanged();180+ };181+176182 async #settle(mutation: MutationHandle): Promise<void> {177183 const generation = ++this.#writeGeneration;178- let localPending = true;179184 this.#trackBatch(mutation);180- this.#diagnostics.pendingLocalWrites += 1;181- this.#diagnosticsChanged();182185 // A store that has not received its first query result stays loading —183186 // a mutation must not present an empty table as ready.184187 this.#set({@@ -187,39 +190,29 @@ export class TableStore<T extends TableRow, Init> {
187190 error: null,188191 });189192 try {190- await mutation.wait({ tier: "local" });191- localPending = false;192- this.#diagnostics.pendingLocalWrites -= 1;193- this.#diagnostics.localWaitCalls += 1;194- this.#diagnosticsChanged();195- if (generation === this.#writeGeneration) {196- this.#set({ durability: "local", error: null });197- }198- if (this.#syncConfigured()) {199- this.#diagnostics.pendingGlobalWrites += 1;200- this.#diagnosticsChanged();201- void mutation.wait({ tier: "global" }).then(202- () => {193+ // Generation guards keep a superseded write's late outcome — success or194+ // rejection — diagnosable without overwriting the newer write's state.195+ await settleDurableWrite(196+ mutation,197+ this.#updateDiagnostics,198+ this.#syncConfigured() ? "background" : "none",199+ {200+ onLocal: () => {201+ if (generation === this.#writeGeneration) {202+ this.#set({ durability: "local", error: null });203+ }204+ },205+ onGlobal: () => {203206 if (generation === this.#writeGeneration) {204207 this.#set({ durability: "global", error: null });205208 }206209 },207- (error) => {208- this.#diagnostics.mutationErrors += 1;209- this.#diagnosticsChanged();210- // Match the success guard: a superseded write's late rejection is211- // diagnosable but must not overwrite the newer write's durability.210+ onGlobalError: (error) => {212211 if (generation === this.#writeGeneration) this.#fail(error);213212 },214- ).finally(() => {215- this.#diagnostics.pendingGlobalWrites -= 1;216- this.#diagnosticsChanged();217- });218- }213+ },214+ );219215 } catch (error) {220- if (localPending) this.#diagnostics.pendingLocalWrites -= 1;221- this.#diagnostics.mutationErrors += 1;222- this.#diagnosticsChanged();223216 this.#fail(error);224217 throw error;225218 }