alexion
6081e3e325
Imported as a linear vendor commit so forge rebases do not replay the upstream root tree. git-subtree-dir: third_party/pi git-subtree-split: 71023aec1546d6d677a65669c5665c0927ce0dd0
504 lines
17 KiB
TypeScript
504 lines
17 KiB
TypeScript
import type { Context } from "@earendil-works/chord";
|
|
import {
|
|
BACKGROUND_CONTEXT,
|
|
createContextKey,
|
|
withAbortSignal,
|
|
withCancel,
|
|
withContextValue,
|
|
} from "@earendil-works/chord/context";
|
|
import { applyImmutable } from "@earendil-works/chord/delta";
|
|
import { defineDoc } from "@earendil-works/pi-durable";
|
|
import { describe, expect, it } from "vitest";
|
|
import { context, documentChanges, flush, openTestSession } from "./session-support.ts";
|
|
|
|
type State = { value: number; items: string[]; retained: { label: string } };
|
|
|
|
const StateDoc = defineDoc<State>({
|
|
kind: "watch.state",
|
|
version: 1,
|
|
scope: "session",
|
|
initial: () => ({ value: 0, items: ["a", "b"], retained: { label: "stable" } }),
|
|
});
|
|
|
|
type Deferred = { readonly promise: Promise<void>; readonly resolve: () => void };
|
|
|
|
function deferred(): Deferred {
|
|
let resolve!: () => void;
|
|
const promise = new Promise<void>((done) => {
|
|
resolve = done;
|
|
});
|
|
return { promise, resolve };
|
|
}
|
|
|
|
async function createState(): Promise<ReturnType<typeof openTestSession>> {
|
|
const harness = openTestSession();
|
|
await harness.session.commit((tx) => tx.doc(StateDoc).then(() => undefined), context);
|
|
return harness;
|
|
}
|
|
|
|
describe("Session document watches", () => {
|
|
it("never creates an absent document", async () => {
|
|
const { session, storage } = openTestSession();
|
|
const commits = storage.commits.length;
|
|
expect(await session.watchDoc(StateDoc, context)).toBeUndefined();
|
|
expect(storage.commits).toHaveLength(commits);
|
|
expect(storage.mintCount).toBe(0);
|
|
});
|
|
|
|
it("keeps the acquisition revision until start and delivers exact committed frames", async () => {
|
|
const { session, publications } = await createState();
|
|
const watch = (await session.watchDoc(StateDoc, context))!;
|
|
const initial = watch.value;
|
|
await session.commit(async (tx) => {
|
|
(await tx.doc(StateDoc)).value = 1;
|
|
}, context);
|
|
await flush();
|
|
const firstPublished = documentChanges(publications.at(-1)!)[0]!;
|
|
await session.commit(async (tx) => {
|
|
(await tx.doc(StateDoc)).value = 2;
|
|
}, context);
|
|
await flush();
|
|
const secondPublished = documentChanges(publications.at(-1)!)[0]!;
|
|
expect(watch.value).toBe(initial);
|
|
|
|
let inline = true;
|
|
const deliveries: Array<{ value: Readonly<State> | null; ops: unknown }> = [];
|
|
watch.start(async (value, ops) => {
|
|
expect(inline).toBe(false);
|
|
expect(watch.value).toBe(value);
|
|
deliveries.push({ value, ops });
|
|
});
|
|
inline = false;
|
|
expect(deliveries).toHaveLength(0);
|
|
await flush();
|
|
expect(deliveries.map(({ value }) => value?.value)).toEqual([1, 2]);
|
|
expect(deliveries[0]!.value).toBe(firstPublished.value);
|
|
expect(deliveries[0]!.ops).toBe(firstPublished.ops);
|
|
expect(deliveries[1]!.value).toBe(secondPublished.value);
|
|
expect(deliveries[1]!.ops).toBe(secondPublished.ops);
|
|
expect(initial?.value).toBe(0);
|
|
await watch.stop();
|
|
});
|
|
|
|
it("serializes callbacks and buffers exact frames committed while one is in flight", async () => {
|
|
const { session } = await createState();
|
|
const watch = (await session.watchDoc(StateDoc, context))!;
|
|
const entered = deferred();
|
|
const release = deferred();
|
|
const values: number[] = [];
|
|
let active = 0;
|
|
let maxActive = 0;
|
|
watch.start(async (value) => {
|
|
active++;
|
|
maxActive = Math.max(maxActive, active);
|
|
values.push(value?.value ?? -1);
|
|
if (values.length === 1) {
|
|
entered.resolve();
|
|
await release.promise;
|
|
}
|
|
active--;
|
|
});
|
|
await session.commit(async (tx) => {
|
|
(await tx.doc(StateDoc)).value = 1;
|
|
}, context);
|
|
await entered.promise;
|
|
for (let value = 2; value <= 20; value++) {
|
|
await session.commit(async (tx) => {
|
|
(await tx.doc(StateDoc)).value = value;
|
|
}, context);
|
|
}
|
|
expect(values).toEqual([1]);
|
|
release.resolve();
|
|
await flush();
|
|
expect(maxActive).toBe(1);
|
|
expect(values).toEqual(Array.from({ length: 20 }, (_, index) => index + 1));
|
|
await watch.stop();
|
|
});
|
|
|
|
it("allows a listener to initiate a later Session commit", async () => {
|
|
const { session } = await createState();
|
|
const watch = (await session.watchDoc(StateDoc, context))!;
|
|
const completed = deferred();
|
|
const values: number[] = [];
|
|
watch.start(async (value) => {
|
|
const current = value?.value ?? -1;
|
|
values.push(current);
|
|
if (current === 1) {
|
|
await session.commit(async (tx) => {
|
|
(await tx.doc(StateDoc)).value = 2;
|
|
}, context);
|
|
} else {
|
|
completed.resolve();
|
|
}
|
|
});
|
|
await session.commit(async (tx) => {
|
|
(await tx.doc(StateDoc)).value = 1;
|
|
}, context);
|
|
await completed.promise;
|
|
expect(values).toEqual([1, 2]);
|
|
await watch.stop();
|
|
});
|
|
|
|
it("collapses 101 pending commits to one root replacement", async () => {
|
|
const { session } = await createState();
|
|
const watch = (await session.watchDoc(StateDoc, context))!;
|
|
for (let value = 1; value <= 101; value++) {
|
|
await session.commit(async (tx) => {
|
|
(await tx.doc(StateDoc)).value = value;
|
|
}, context);
|
|
}
|
|
const deliveries: Array<{ value: number; ops: unknown }> = [];
|
|
watch.start(async (value, ops) => {
|
|
deliveries.push({ value: value?.value ?? -1, ops });
|
|
});
|
|
await flush();
|
|
expect(deliveries).toEqual([{ value: 101, ops: [["r", watch.value]] }]);
|
|
await watch.stop();
|
|
});
|
|
|
|
it("never folds the in-flight frame into an overflow reset", async () => {
|
|
const { session } = await createState();
|
|
const watch = (await session.watchDoc(StateDoc, context))!;
|
|
const entered = deferred();
|
|
const release = deferred();
|
|
const deliveries: Array<{ value: number; ops: unknown }> = [];
|
|
watch.start(async (value, ops) => {
|
|
deliveries.push({ value: value?.value ?? -1, ops });
|
|
if (deliveries.length === 1) {
|
|
entered.resolve();
|
|
await release.promise;
|
|
}
|
|
});
|
|
await session.commit(async (tx) => {
|
|
(await tx.doc(StateDoc)).value = 1;
|
|
}, context);
|
|
await entered.promise;
|
|
for (let value = 2; value <= 102; value++) {
|
|
await session.commit(async (tx) => {
|
|
(await tx.doc(StateDoc)).value = value;
|
|
}, context);
|
|
}
|
|
release.resolve();
|
|
await flush();
|
|
expect(deliveries[0]!.value).toBe(1);
|
|
expect(deliveries[1]).toEqual({ value: 102, ops: [["r", watch.value]] });
|
|
expect(deliveries).toHaveLength(2);
|
|
await watch.stop();
|
|
});
|
|
|
|
it("folds retirement into an overflow reset and then closes", async () => {
|
|
const { session } = await createState();
|
|
const watch = (await session.watchDoc(StateDoc, context))!;
|
|
for (let value = 1; value <= 100; value++) {
|
|
await session.commit(async (tx) => {
|
|
(await tx.doc(StateDoc)).value = value;
|
|
}, context);
|
|
}
|
|
await session.commit((tx) => tx.retireDoc(StateDoc), context);
|
|
const deliveries: Array<{ value: Readonly<State> | null; ops: unknown }> = [];
|
|
watch.start(async (value, ops) => {
|
|
deliveries.push({ value, ops });
|
|
});
|
|
expect(await watch.closed).toEqual({ reason: "retired" });
|
|
expect(deliveries).toEqual([{ value: null, ops: [["r", null]] }]);
|
|
});
|
|
|
|
it("delivers replayable structural no-op commits instead of suppressing them", async () => {
|
|
const { session } = await createState();
|
|
const watch = (await session.watchDoc(StateDoc, context))!;
|
|
const initial = watch.value;
|
|
await session.commit(async (tx) => {
|
|
const state = await tx.doc(StateDoc);
|
|
const first = state.items.shift()!;
|
|
state.items.unshift(first);
|
|
}, context);
|
|
const newest = (await session.snapshot(StateDoc, context))!;
|
|
expect(newest).not.toBe(initial);
|
|
expect(newest).toEqual(initial);
|
|
const batches: unknown[] = [];
|
|
watch.start(async (value, ops) => {
|
|
expect(value).toBe(newest);
|
|
batches.push(ops);
|
|
});
|
|
await flush();
|
|
expect(batches).toHaveLength(1);
|
|
expect((batches[0] as readonly unknown[]).length).toBeGreaterThan(0);
|
|
await watch.stop();
|
|
});
|
|
|
|
it("preserves commit Context values without inheriting producer cancellation", async () => {
|
|
const { session } = await createState();
|
|
const watch = (await session.watchDoc(StateDoc, context))!;
|
|
const key = createContextKey<string>("watch-test");
|
|
const parentController = new AbortController();
|
|
const commitContext = withContextValue(
|
|
key,
|
|
"newest-commit",
|
|
withAbortSignal(parentController.signal, BACKGROUND_CONTEXT),
|
|
);
|
|
const entered = deferred();
|
|
const release = deferred();
|
|
let deliveryContext: Context | undefined;
|
|
watch.start(async (_value, _ops, delivered) => {
|
|
deliveryContext = delivered;
|
|
entered.resolve();
|
|
await release.promise;
|
|
});
|
|
await session.commit(async (tx) => {
|
|
(await tx.doc(StateDoc)).value = 1;
|
|
}, commitContext);
|
|
await entered.promise;
|
|
expect(deliveryContext!.value(key)).toBe("newest-commit");
|
|
expect(deliveryContext!.abortSignal).toBeUndefined();
|
|
parentController.abort("caller finished");
|
|
const stopped = watch.stop();
|
|
expect(await stopped).toEqual({ reason: "stopped" });
|
|
expect(deliveryContext!.abortSignal).toBeUndefined();
|
|
release.resolve();
|
|
});
|
|
|
|
it("keeps earlier immutable revisions stable", async () => {
|
|
const { session } = await createState();
|
|
const watch = (await session.watchDoc(StateDoc, context))!;
|
|
const initial = watch.value;
|
|
let delivered: Readonly<State> | null | undefined;
|
|
watch.start(async (value) => {
|
|
delivered = value;
|
|
});
|
|
await session.commit(async (tx) => {
|
|
(await tx.doc(StateDoc)).value = 1;
|
|
}, context);
|
|
await flush();
|
|
expect(delivered).toBe(watch.value);
|
|
expect(delivered).not.toBe(initial);
|
|
expect(delivered?.retained).toBe(initial?.retained);
|
|
expect(initial?.value).toBe(0);
|
|
await watch.stop();
|
|
});
|
|
|
|
it("delivers retirement and does not follow recreation", async () => {
|
|
const { session } = await createState();
|
|
const oldWatch = (await session.watchDoc(StateDoc, context))!;
|
|
const values: Array<Readonly<State> | null> = [];
|
|
oldWatch.start(async (value) => {
|
|
values.push(value);
|
|
});
|
|
await session.commit(async (tx) => {
|
|
await tx.retireDoc(StateDoc);
|
|
(await tx.doc(StateDoc)).value = 10;
|
|
}, context);
|
|
expect(await oldWatch.closed).toEqual({ reason: "retired" });
|
|
expect(values).toEqual([null]);
|
|
expect(oldWatch.value).toBeNull();
|
|
|
|
const replacement = (await session.watchDoc(StateDoc, context))!;
|
|
expect(replacement.value?.value).toBe(10);
|
|
await session.commit(async (tx) => {
|
|
(await tx.doc(StateDoc)).value = 11;
|
|
}, context);
|
|
await flush();
|
|
expect(oldWatch.value).toBeNull();
|
|
await replacement.stop();
|
|
});
|
|
|
|
it("Session close discards retirement buffered before start", async () => {
|
|
const { session } = await createState();
|
|
const watch = (await session.watchDoc(StateDoc, context))!;
|
|
const baseline = watch.value;
|
|
await session.commit((tx) => tx.retireDoc(StateDoc), context);
|
|
await session.close(context);
|
|
expect(await watch.closed).toEqual({ reason: "session_closed" });
|
|
expect(watch.value).toBe(baseline);
|
|
expect(() => watch.start(async () => undefined)).toThrow("stopped");
|
|
});
|
|
|
|
it("Session close discards retirement behind an in-flight callback", async () => {
|
|
const { session } = await createState();
|
|
const watch = (await session.watchDoc(StateDoc, context))!;
|
|
const entered = deferred();
|
|
const release = deferred();
|
|
const values: Array<number | null> = [];
|
|
watch.start(async (value) => {
|
|
values.push(value?.value ?? null);
|
|
entered.resolve();
|
|
await release.promise;
|
|
});
|
|
await session.commit(async (tx) => {
|
|
(await tx.doc(StateDoc)).value = 1;
|
|
}, context);
|
|
await entered.promise;
|
|
await session.commit((tx) => tx.retireDoc(StateDoc), context);
|
|
await session.close(context);
|
|
expect(await watch.closed).toEqual({ reason: "session_closed" });
|
|
release.resolve();
|
|
await flush();
|
|
expect(values).toEqual([1]);
|
|
});
|
|
|
|
it("supports idempotent stop and rejects repeated or late start", async () => {
|
|
const { session } = await createState();
|
|
const started = (await session.watchDoc(StateDoc, context))!;
|
|
started.start(async () => undefined);
|
|
expect(() => started.start(async () => undefined)).toThrow("already started");
|
|
const first = started.stop();
|
|
const second = started.stop();
|
|
expect(second).toBe(first);
|
|
expect(await first).toEqual({ reason: "stopped" });
|
|
|
|
const stopped = (await session.watchDoc(StateDoc, context))!;
|
|
await stopped.stop();
|
|
expect(() => stopped.start(async () => undefined)).toThrow("stopped");
|
|
});
|
|
|
|
it("cancels acquisition without leaking a registered watch", async () => {
|
|
const { session, storage } = await createState();
|
|
await session.unloadDocuments();
|
|
const gate = storage.holdFindDocument();
|
|
const child = withCancel(context);
|
|
const acquisition = session.watchDoc(StateDoc, child.context);
|
|
await gate.entered;
|
|
child.cancel(new Error("cancel acquisition"));
|
|
gate.release();
|
|
await expect(acquisition).rejects.toThrow("cancel acquisition");
|
|
await session.close(context);
|
|
});
|
|
|
|
it("cancels future delivery without aborting an in-flight callback", async () => {
|
|
const { session } = await createState();
|
|
const child = withCancel(context);
|
|
const watch = (await session.watchDoc(StateDoc, child.context))!;
|
|
const entered = deferred();
|
|
const release = deferred();
|
|
let deliveryContext: Context | undefined;
|
|
watch.start(async (_value, _ops, delivered) => {
|
|
deliveryContext = delivered;
|
|
entered.resolve();
|
|
await release.promise;
|
|
});
|
|
await session.commit(async (tx) => {
|
|
(await tx.doc(StateDoc)).value = 1;
|
|
}, context);
|
|
await entered.promise;
|
|
child.cancel();
|
|
expect(await watch.closed).toEqual({ reason: "cancelled" });
|
|
expect(deliveryContext!.abortSignal).toBeUndefined();
|
|
release.resolve();
|
|
});
|
|
|
|
it("Session close stops future delivery without joining an in-flight callback", async () => {
|
|
const { session } = await createState();
|
|
const watch = (await session.watchDoc(StateDoc, context))!;
|
|
const entered = deferred();
|
|
const release = deferred();
|
|
let deliveryContext: Context | undefined;
|
|
watch.start(async (_value, _ops, delivered) => {
|
|
deliveryContext = delivered;
|
|
entered.resolve();
|
|
await release.promise;
|
|
});
|
|
await session.commit(async (tx) => {
|
|
(await tx.doc(StateDoc)).value = 1;
|
|
}, context);
|
|
await entered.promise;
|
|
await session.close(context);
|
|
expect(deliveryContext!.abortSignal).toBeUndefined();
|
|
expect(await watch.closed).toEqual({ reason: "session_closed" });
|
|
release.resolve();
|
|
});
|
|
|
|
it("settles listener failure on only the affected watch", async () => {
|
|
const { session } = await createState();
|
|
const failed = (await session.watchDoc(StateDoc, context))!;
|
|
const healthy = (await session.watchDoc(StateDoc, context))!;
|
|
failed.start(async () => {
|
|
throw new Error("listener failed");
|
|
});
|
|
let healthyCalls = 0;
|
|
healthy.start(async () => {
|
|
healthyCalls++;
|
|
});
|
|
await session.commit(async (tx) => {
|
|
(await tx.doc(StateDoc)).value = 1;
|
|
}, context);
|
|
const end = await failed.closed;
|
|
expect(end.reason).toBe("listener_error");
|
|
if (end.reason === "listener_error") expect(end.error.message).toBe("listener failed");
|
|
await flush();
|
|
expect(healthyCalls).toBe(1);
|
|
await healthy.stop();
|
|
});
|
|
|
|
it("hydrates migration without writing and observes the later exact edit", async () => {
|
|
const Old = defineDoc<{ value: number }>({
|
|
kind: "watch.migration",
|
|
version: 1,
|
|
scope: "session",
|
|
initial: () => ({ value: 3 }),
|
|
});
|
|
const Current = defineDoc<{ value: number; migrated: boolean }>({
|
|
kind: "watch.migration",
|
|
version: 2,
|
|
scope: "session",
|
|
initial: () => ({ value: 0, migrated: false }),
|
|
migrate: (value) => ({ value: value.value as number, migrated: true }),
|
|
});
|
|
const { session, storage } = openTestSession();
|
|
await session.commit((tx) => tx.doc(Old).then(() => undefined), context);
|
|
await session.unloadDocuments();
|
|
const commits = storage.commits.length;
|
|
const watch = (await session.watchDoc(Current, context))!;
|
|
expect(watch.value).toEqual({ value: 3, migrated: true });
|
|
expect(storage.commits).toHaveLength(commits);
|
|
const values: number[] = [];
|
|
watch.start(async (value) => {
|
|
values.push(value?.value ?? -1);
|
|
});
|
|
await session.commit((tx) => tx.doc(Current).then(() => undefined), context);
|
|
await flush();
|
|
expect(values).toEqual([]);
|
|
await session.commit(async (tx) => {
|
|
(await tx.doc(Current)).value = 4;
|
|
}, context);
|
|
await flush();
|
|
expect(values).toEqual([4]);
|
|
await watch.stop();
|
|
});
|
|
|
|
it("can replay every delivered exact operation batch from the acquisition revision", async () => {
|
|
const { session } = await createState();
|
|
const watch = (await session.watchDoc(StateDoc, context))!;
|
|
let replica: Readonly<State> | null = watch.value;
|
|
watch.start(async (value, ops) => {
|
|
replica = applyImmutable(replica, ops);
|
|
expect(replica).toEqual(value);
|
|
});
|
|
await session.commit(async (tx) => {
|
|
const state = await tx.doc(StateDoc);
|
|
state.value = 3;
|
|
state.items.splice(0, 1);
|
|
}, context);
|
|
await flush();
|
|
expect(replica).toEqual(watch.value);
|
|
await watch.stop();
|
|
});
|
|
|
|
it("continues after the tracker cache unloads", async () => {
|
|
const { session, storage } = await createState();
|
|
const watch = (await session.watchDoc(StateDoc, context))!;
|
|
const baseline = watch.value;
|
|
const reads = storage.documentReadCount;
|
|
await session.unloadDocuments();
|
|
const reloaded = await session.snapshot(StateDoc, context);
|
|
expect(reloaded).toEqual(baseline);
|
|
expect(reloaded).not.toBe(baseline);
|
|
expect(storage.documentReadCount).toBeGreaterThan(reads);
|
|
watch.start(async () => undefined);
|
|
await session.commit(async (tx) => {
|
|
(await tx.doc(StateDoc)).value = 7;
|
|
}, context);
|
|
await flush();
|
|
expect(watch.value?.value).toBe(7);
|
|
await watch.stop();
|
|
});
|
|
});
|