mirror of
https://github.com/badlogic/pi-mono.git
synced 2026-11-05 23:37:25 +00:00
a5d209afd0
The pure swap is gone: a reload restarts every live object of a changed plugin (documents and tables kept). Calls running at the commit and every call they cause (the ctx call chain; an old flag on call from trusted links) run on the open old attempt, which seals once its work drained and its successor's recorded requests are ready, bounded by reloadDrainTimeout (5 s); then the successor's factory runs. New work waits at the gate; same-host consumers see the draining object live; a reload cycle fails at once with unavailable/reload_cycle. Example 64, design section 21 rewritten.
349 lines
13 KiB
TypeScript
349 lines
13 KiB
TypeScript
import { InMemoryTelemetryContext } from "@earendil-works/pi-telemetry";
|
|
import { describe, expect, it } from "vitest";
|
|
import { BACKGROUND_CONTEXT as context, withAbortSignal, withTelemetry } from "../../src/context/index.ts";
|
|
import { int, local, method, remote, str } from "../../src/contracts/index.ts";
|
|
import { withLinkedAbortSignal } from "../../src/host/ctx.ts";
|
|
import type { ChordHost } from "../../src/host/host.ts";
|
|
import type { Ctx, Host, Reference } from "../../src/host/index.ts";
|
|
import { CALL_CHAIN } from "../../src/host/old-work.ts";
|
|
import { constant } from "../../src/state/index.ts";
|
|
import {
|
|
flushed,
|
|
incarnationOf,
|
|
linked,
|
|
loaded,
|
|
rawParent,
|
|
rejection,
|
|
Server,
|
|
type TestPrincipal,
|
|
} from "../support.ts";
|
|
|
|
const Item = local("app.item", { version: 1, params: { id: int() }, members: { id: int() } });
|
|
const Desk = local("app.desk", {
|
|
version: 1,
|
|
params: {},
|
|
members: { use: method(undefined, int()), leak: method(), late: method(), who: method(undefined, str()) },
|
|
});
|
|
|
|
/** A desk whose handlers use `ctx.get`; what they got and their contexts are recorded. */
|
|
async function desk() {
|
|
const seen: { ctx?: Ctx; item?: Reference<typeof Item>; late?: Promise<unknown>; factories: number } = {
|
|
factories: 0,
|
|
};
|
|
const { host } = await loaded((h) => {
|
|
h.serve(Item, ({ id }) => {
|
|
seen.factories++;
|
|
return { id: constant(id) };
|
|
});
|
|
h.serve(Desk, () => ({
|
|
use: async (ctx) => {
|
|
const item = await ctx.get(Item, { id: 1 });
|
|
return item.id.value;
|
|
},
|
|
leak: async (ctx) => {
|
|
seen.ctx = ctx;
|
|
seen.item = await ctx.get(Item, { id: 2 });
|
|
return null;
|
|
},
|
|
late: async (ctx) => {
|
|
seen.late = ctx.get(Item, { id: 3 }).catch((error: unknown) => error);
|
|
return null;
|
|
},
|
|
who: async (ctx) => `${ctx.principal?.kind ?? "none"} ${ctx.incarnation}`,
|
|
}));
|
|
});
|
|
return { host, seen, desk: await host.get(Desk, {}, context) };
|
|
}
|
|
|
|
describe("ctx", () => {
|
|
it("ctx.get references are released when the handler settles", async () => {
|
|
const { desk: ref, seen } = await desk();
|
|
expect(await ref.use(context)).toBe(1);
|
|
await ref.leak(context);
|
|
expect(() => seen.item!.id.value).toThrow("app.item: released");
|
|
// The object ended with its last reference: a new get creates it anew.
|
|
await ref.use(context);
|
|
expect(seen.factories).toBe(3);
|
|
});
|
|
|
|
it("ctx.get after the handler settled throws released", async () => {
|
|
const { desk: ref, seen } = await desk();
|
|
await ref.leak(context);
|
|
expect(await rejection(seen.ctx!.get(Item, { id: 1 }))).toEqual({
|
|
code: "released",
|
|
reason: "settled",
|
|
message: "app.item: ctx.get after its call settled",
|
|
});
|
|
});
|
|
|
|
it("ctx.get of a handler whose call was aborted checks the signal first: nothing is created", async () => {
|
|
let factories = 0;
|
|
let got: Promise<unknown> | undefined;
|
|
const gate = Promise.withResolvers<void>();
|
|
const { host } = await loaded((h) => {
|
|
h.serve(Item, ({ id }) => {
|
|
factories++;
|
|
return { id: constant(id) };
|
|
});
|
|
h.serve(Desk, () => ({
|
|
use: async (ctx) => {
|
|
await gate.promise;
|
|
got = ctx.get(Item, { id: 1 }).catch((error: unknown) => error);
|
|
return 0;
|
|
},
|
|
leak: async () => null,
|
|
late: async () => null,
|
|
who: async () => "",
|
|
}));
|
|
});
|
|
const ref = await host.get(Desk, {}, context);
|
|
const caller = new AbortController();
|
|
const call = ref.use(withAbortSignal(caller.signal, context)).catch(() => {});
|
|
await flushed();
|
|
caller.abort(new Error("caller gave up"));
|
|
gate.resolve();
|
|
await call;
|
|
await flushed();
|
|
expect(await got).toEqual(new Error("caller gave up"));
|
|
expect(factories).toBe(0);
|
|
});
|
|
|
|
it("a ctx.get finishing after its handler settled is released at once (race 11)", async () => {
|
|
const { desk: ref, seen } = await desk();
|
|
await ref.late(context);
|
|
expect(await seen.late).toMatchObject({
|
|
code: "released",
|
|
reason: "settled",
|
|
message: "app.item: ctx.get after its call settled",
|
|
});
|
|
await flushed();
|
|
// The late reference's object ended with it.
|
|
expect(seen.factories).toBe(1);
|
|
});
|
|
|
|
it("a ctx.get still waiting when its handler settles gives up: its reference is released at once", async () => {
|
|
const Far = remote("app.far", {
|
|
version: 1,
|
|
params: {},
|
|
hostAddress: () => Server(),
|
|
variants: { v: { state: { n: int() } } },
|
|
});
|
|
const Starter = local("app.starter", { version: 1, params: {}, members: { start: method() } });
|
|
let pending: Promise<unknown> | undefined;
|
|
const { host } = await loaded((h) =>
|
|
h.serve(Starter, () => ({
|
|
start: async (ctx) => {
|
|
pending = ctx.get(Far, {}).catch((error: unknown) => error);
|
|
return null;
|
|
},
|
|
})),
|
|
);
|
|
// A host of any principal type takes a parent link.
|
|
const parent = await rawParent(host as Host<TestPrincipal>);
|
|
await (await host.get(Starter, {}, context)).start(context);
|
|
const { hold } = await parent.next("hold");
|
|
expect(await parent.next("release")).toMatchObject({ hold });
|
|
expect(await pending).toMatchObject({
|
|
code: "released",
|
|
reason: "settled",
|
|
message: "app.far: ctx.get after its call settled",
|
|
});
|
|
});
|
|
|
|
it("a local handler's ctx.principal is the host's, its incarnation this host's", async () => {
|
|
const { desk: ref, host } = await desk();
|
|
expect(await ref.who(context)).toBe(`none ${(host as ChordHost).incarnation}`);
|
|
});
|
|
});
|
|
|
|
describe('the call chain (decision "Reload restarts every affected object": old work)', () => {
|
|
it("a method's ctx is its own chain node, above it the chain of the call that made it; a root's has none", async () => {
|
|
const Inner = local("app.inner", { version: 1, params: {}, members: { run: method() } });
|
|
const Outer = local("app.outer", { version: 1, params: {}, members: { run: method() } });
|
|
const seen: { inner?: Ctx; outer?: Ctx; root?: Ctx } = {};
|
|
const { host } = await loaded((h) => {
|
|
h.serve(Inner, () => ({
|
|
run: async (ctx) => {
|
|
seen.inner = ctx;
|
|
return null;
|
|
},
|
|
}));
|
|
h.serve(Outer, () => ({
|
|
run: async (ctx) => {
|
|
seen.outer = ctx;
|
|
await (await ctx.get(Inner, {})).run(ctx);
|
|
return null;
|
|
},
|
|
}));
|
|
});
|
|
await host.root(async (root) => {
|
|
seen.root = root;
|
|
await (await root.get(Outer, {})).run(root);
|
|
}, context);
|
|
const inner = seen.inner!.value(CALL_CHAIN);
|
|
expect(inner).toBe(seen.inner);
|
|
expect(inner!.up()).toBe(seen.outer);
|
|
expect(inner!.up()!.up()).toBeUndefined();
|
|
expect(seen.root!.value(CALL_CHAIN)).toBeUndefined();
|
|
await host.close(context);
|
|
});
|
|
});
|
|
|
|
describe("local handlers act as the host (§4, audit)", () => {
|
|
it("a local object called from a remote handler's ctx.get runs with this host's principal and incarnation", async () => {
|
|
const Store = local("app.local-store", { version: 1, params: {}, members: { who: method(undefined, str()) } });
|
|
const Gate = remote("app.gate", {
|
|
version: 1,
|
|
params: {},
|
|
hostAddress: () => Server(),
|
|
variants: { v: { methods: { who: method(undefined, str()), through: method(undefined, str()) } } },
|
|
});
|
|
const who = (ctx: Ctx<TestPrincipal>) => `${ctx.principal?.kind ?? "none"} ${ctx.incarnation}`;
|
|
const { server, client } = await linked((h) => {
|
|
h.serve(Store, () => ({ who: async (ctx) => who(ctx) }));
|
|
h.serve(Gate, () => ({
|
|
variant: "v",
|
|
members: {
|
|
who: async (ctx) => who(ctx),
|
|
through: async (ctx) => (await ctx.get(Store, {})).who(ctx),
|
|
},
|
|
}));
|
|
});
|
|
const gate = await client.get(Gate, {}, context);
|
|
// The remote handler acts as its caller; the local object it reaches acts as this host, both halves.
|
|
expect(await gate.who(context)).toBe(`owner ${incarnationOf(client)}`);
|
|
expect(await gate.through(context)).toBe(`server ${incarnationOf(server)}`);
|
|
});
|
|
});
|
|
|
|
describe("linked abort signals", () => {
|
|
it("abort with the caller's signal or the call's, the caller's reason first", () => {
|
|
for (const first of ["caller", "call"] as const) {
|
|
const [caller, call] = [new AbortController(), new AbortController()];
|
|
const { context: linked } = withLinkedAbortSignal(call.signal, withAbortSignal(caller.signal, context));
|
|
(first === "caller" ? caller : call).abort(first);
|
|
expect(linked.abortSignal?.reason).toBe(first);
|
|
}
|
|
const [caller, call] = [new AbortController(), new AbortController()];
|
|
call.abort("call");
|
|
caller.abort("caller");
|
|
const { context: linked } = withLinkedAbortSignal(call.signal, withAbortSignal(caller.signal, context));
|
|
expect(linked.abortSignal?.reason).toBe("caller");
|
|
});
|
|
|
|
it("detach from both sources on release: a later abort reaches nothing", () => {
|
|
const [caller, call] = [new AbortController(), new AbortController()];
|
|
const linked = withLinkedAbortSignal(call.signal, withAbortSignal(caller.signal, context));
|
|
linked.release();
|
|
caller.abort("late");
|
|
call.abort("late");
|
|
expect(linked.context.abortSignal?.aborted).toBe(false);
|
|
});
|
|
|
|
it("is the call's own signal without a caller's", () => {
|
|
const call = new AbortController();
|
|
const linked = withLinkedAbortSignal(call.signal, context);
|
|
expect(linked.context.abortSignal).toBe(call.signal);
|
|
linked.release();
|
|
});
|
|
});
|
|
|
|
describe("ctx.trace (C11, slice 9)", () => {
|
|
const Traced = local("app.traced", {
|
|
version: 1,
|
|
params: {},
|
|
members: { work: method(undefined, int()), failing: method() },
|
|
});
|
|
|
|
/** A host recording to `telemetry`, serving `Traced`, whose `work` runs two child spans and gets an item in one. */
|
|
async function traced(telemetry: InMemoryTelemetryContext | undefined) {
|
|
const seen: { inner?: Ctx; outer?: Ctx; item?: Reference<typeof Item> } = {};
|
|
const { host } = await loaded(
|
|
(h) => {
|
|
h.serve(Item, ({ id }) => ({ id: constant(id) }));
|
|
h.serve(Traced, () => ({
|
|
work: async (ctx) => {
|
|
seen.outer = ctx;
|
|
return ctx.trace("app.step", { n: 1 }, async (inner) => {
|
|
seen.inner = inner;
|
|
seen.item = await inner.get(Item, { id: 3 });
|
|
return inner.trace("app.deeper", {}, async () => seen.item!.id.value);
|
|
});
|
|
},
|
|
failing: async (ctx) => {
|
|
await ctx.trace("app.fails", {}, async () => {
|
|
throw new Error("broke");
|
|
});
|
|
return null;
|
|
},
|
|
}));
|
|
},
|
|
{ telemetry },
|
|
);
|
|
return { host, seen };
|
|
}
|
|
|
|
it("runs fn in a child span of the handler's, with a child ctx: the same principal, signal and references", async () => {
|
|
const telemetry = new InMemoryTelemetryContext();
|
|
const { host, seen } = await traced(telemetry);
|
|
const ref = await host.get(Traced, {}, context);
|
|
await telemetry.startSpan({ name: "caller" }, async (span) => {
|
|
expect(await ref.work(withTelemetry(span, context))).toBe(3);
|
|
});
|
|
const spans = telemetry.getSpans();
|
|
const byName = (name: string) => spans.find((span) => span.name === name)!;
|
|
expect(byName("app.step").parentId).toBe(byName("chord.method").id);
|
|
expect(byName("app.step").attributes).toEqual({ "chord.host": "client", n: 1 });
|
|
expect(byName("app.deeper").parentId).toBe(byName("app.step").id);
|
|
expect(seen.inner).not.toBe(seen.outer);
|
|
expect(seen.inner!.principal).toBe(seen.outer!.principal);
|
|
expect(seen.inner!.incarnation).toBe(seen.outer!.incarnation);
|
|
// The child's references are the handler's: released when the handler settled.
|
|
expect(seen.item!.state.value.kind).toBe("ended");
|
|
await expect(seen.inner!.get(Item, { id: 3 })).rejects.toMatchObject({ code: "released" });
|
|
});
|
|
|
|
it("marks its span failed (`trace_failed`) and rethrows", async () => {
|
|
const telemetry = new InMemoryTelemetryContext();
|
|
const { host } = await traced(telemetry);
|
|
const ref = await host.get(Traced, {}, context);
|
|
await telemetry.startSpan({ name: "caller" }, async (span) => {
|
|
await expect(ref.failing(withTelemetry(span, context))).rejects.toThrow("broke");
|
|
});
|
|
const failed = telemetry.getSpans().find(({ name }) => name === "app.fails")!;
|
|
expect(failed.status).toEqual({ status: "error", error: { name: "trace_failed", message: "broke" } });
|
|
});
|
|
|
|
it("rejects, never throws, when fn throws synchronously, recorded or not (review L5)", async () => {
|
|
for (const telemetry of [new InMemoryTelemetryContext(), undefined]) {
|
|
const { host } = await loaded(
|
|
(h) =>
|
|
h.serve(Traced, () => ({
|
|
work: async (ctx) => {
|
|
const traced = ctx.trace("app.sync", {}, (() => {
|
|
throw new Error("sync");
|
|
}) as () => Promise<number>);
|
|
return traced.catch(() => 7);
|
|
},
|
|
failing: async () => null,
|
|
})),
|
|
{ telemetry },
|
|
);
|
|
const ref = await host.get(Traced, {}, context);
|
|
const work = (caller: typeof context) => ref.work(caller);
|
|
const result =
|
|
telemetry === undefined
|
|
? await work(context)
|
|
: await telemetry.startSpan({ name: "caller" }, (span) => work(withTelemetry(span, context)));
|
|
expect(result).toBe(7);
|
|
if (telemetry !== undefined) expect(telemetry.getSpans().map(({ name }) => name)).toContain("app.sync");
|
|
}
|
|
});
|
|
|
|
it("gives fn the handler's own ctx when nothing records", async () => {
|
|
const { host, seen } = await traced(undefined);
|
|
const ref = await host.get(Traced, {}, context);
|
|
expect(await ref.work(context)).toBe(3);
|
|
expect(seen.inner).toBe(seen.outer);
|
|
});
|
|
});
|