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.
1853 lines
72 KiB
TypeScript
1853 lines
72 KiB
TypeScript
import { InMemoryTelemetryContext, type SpanAttributes } from "@earendil-works/pi-telemetry";
|
|
import { describe, expect, it, vi } from "vitest";
|
|
import { BACKGROUND_CONTEXT as context, withAbortSignal, withCancel, withTelemetry } from "../../src/context/index.ts";
|
|
import { evolve, hostKind, int, local, method, obj, remote, str } from "../../src/contracts/index.ts";
|
|
import { broken } from "../../src/data/index.ts";
|
|
import { type Ctx, createHost, defineFacet, type FacetHost, type HostError, type Self } from "../../src/host/index.ts";
|
|
import type { ChordHost } from "../../src/host/internal.ts";
|
|
import { memoryLinkPair } from "../../src/link/index.ts";
|
|
import { jsonPart } from "../../src/link/internal.ts";
|
|
import { defineRegistry } from "../../src/registries/index.ts";
|
|
import { openNodeSqliteDatabase } from "../../src/sqlite/node.ts";
|
|
import { constant } from "../../src/state/index.ts";
|
|
import { sqliteBackend } from "../../src/store/backend.ts";
|
|
import type { StagedWrite } from "../../src/store/step.ts";
|
|
import { type ChordStore, openStore } from "../../src/store/store.ts";
|
|
import {
|
|
failures,
|
|
fakeTimers,
|
|
faultyStore,
|
|
flushed,
|
|
forEachCodec,
|
|
linked,
|
|
loaded,
|
|
Notes,
|
|
PlacedNotes,
|
|
partValue,
|
|
RawPeer,
|
|
rejection,
|
|
relayed,
|
|
SERVER_ADDRESS,
|
|
Server,
|
|
Session,
|
|
serveNotes,
|
|
statesOf,
|
|
type TestPrincipal,
|
|
testCodec,
|
|
transported,
|
|
} from "../support.ts";
|
|
|
|
const Info = local("app.info", { version: 1, params: {}, members: { title: str() } });
|
|
const Counter = local("app.counter", { version: 1, params: {}, members: { count: int(), bump: method() } });
|
|
const Page = local("app.page", { version: 1, params: { id: int() }, members: { id: int() } });
|
|
const Svc = local("app.svc", { version: 1, params: { id: int() }, members: { n: int() } });
|
|
const Core = local("app.core", { version: 1, params: {}, members: { n: int() } });
|
|
|
|
const noErrors = { onError: () => {} };
|
|
|
|
/** Serves `Svc`: one native whose open waits for `gate` and whose close records its value and reason in `closes`. */
|
|
function serveGated(h: FacetHost, gate: Promise<void>, closes: string[]): void {
|
|
h.serve(Svc, ({ id }, self) => {
|
|
self.open(
|
|
async () => {
|
|
await gate;
|
|
return `svc ${id}`;
|
|
},
|
|
(value, reason) => void closes.push(`${value} ${reason.kind}`),
|
|
);
|
|
return { n: constant(id) };
|
|
});
|
|
}
|
|
|
|
describe("load", () => {
|
|
it("runs every setup and activates the types with empty params before it resolves", async () => {
|
|
const factories = { info: vi.fn(() => ({ title: constant("t") })), page: vi.fn() };
|
|
const facet = defineFacet((host) => {
|
|
host.serve(Info, factories.info);
|
|
host.serve(Page, factories.page);
|
|
});
|
|
const host = createHost(noErrors);
|
|
await host.load([{ name: "@app/info", facet }], context);
|
|
expect(factories.info).toHaveBeenCalledTimes(1);
|
|
expect(factories.page).not.toHaveBeenCalled();
|
|
await host.close(context);
|
|
});
|
|
|
|
it("leaves out and reports duplicate plugin names and double serves (the later); refuses a second load", async () => {
|
|
const empty = defineFacet(() => {});
|
|
const info = defineFacet((host) => host.serve(Info, () => ({ title: constant("t") })));
|
|
const loaded = async (facets: { readonly name: string; readonly facet: typeof empty }[]) => {
|
|
const host = createHost(noErrors);
|
|
const { failed } = await host.load(facets, context);
|
|
await host.close(context);
|
|
return failed.map(({ plugin, error }) => [plugin, (error as Error).message]);
|
|
};
|
|
const twice = [
|
|
{ name: "a", facet: empty },
|
|
{ name: "a", facet: empty },
|
|
];
|
|
expect(await loaded(twice)).toEqual([["a", 'plugin "a" is loaded twice']]);
|
|
const throwing = defineFacet(() => {
|
|
throw new Error("first failed");
|
|
});
|
|
// The first one's setup failing does not make the second one's name free (Sol).
|
|
const failing = createHost(noErrors);
|
|
const report = await failing.load(
|
|
[
|
|
{ name: "a", facet: throwing },
|
|
{ name: "a", facet: info },
|
|
],
|
|
context,
|
|
);
|
|
expect(report.failed.map(({ plugin, error }) => [plugin, (error as Error).message])).toEqual([
|
|
["a", 'plugin "a": first failed'],
|
|
["a", 'plugin "a" is loaded twice'],
|
|
]);
|
|
expect(failing.hold(Info, {}, context).state.value).toMatchObject({ kind: "failed" });
|
|
await failing.close(context);
|
|
const served = [
|
|
{ name: "a", facet: info },
|
|
{ name: "b", facet: info },
|
|
];
|
|
expect(await loaded(served)).toEqual([["b", 'plugin "b": app.info is served twice (plugin "a")']]);
|
|
const host = createHost(noErrors);
|
|
await host.load([], context);
|
|
await expect(host.load([], context)).rejects.toThrow("load runs once per host");
|
|
});
|
|
|
|
it("throws for an invalid plugin name before it changes anything; the host loads afterwards (decision (d))", async () => {
|
|
const empty = defineFacet(() => {});
|
|
const info = defineFacet((h) => h.serve(Info, () => ({ title: constant("t") })));
|
|
const host = createHost(noErrors);
|
|
await expect(
|
|
host.load(
|
|
[
|
|
{ name: "@app/info", facet: info },
|
|
{ name: "tui:core", facet: empty },
|
|
],
|
|
context,
|
|
),
|
|
).rejects.toThrow('"tui:core" is not a valid plugin name');
|
|
// Nothing was staged or published: the same host loads the fixed set.
|
|
expect(await host.load([{ name: "@app/info", facet: info }], context)).toEqual({ failed: [] });
|
|
expect(() => host.reload([{ name: "Bad Name", facet: empty }], context)).toThrow(
|
|
'"Bad Name" is not a valid plugin name',
|
|
);
|
|
// The reload changed nothing: the loaded plugin still serves.
|
|
expect((await host.get(Info, {}, context)).title.value).toBe("t");
|
|
await host.close(context);
|
|
});
|
|
|
|
it("is called first: get, hold, registry and a root's ctx.get throw not_ready before it (decision (a))", async () => {
|
|
const Names = defineRegistry<string, readonly string[]>("app.names", { version: 1 });
|
|
const host = createHost(noErrors);
|
|
const fix = "the host has not loaded; call host.load first (a host with nothing to load calls load([]))";
|
|
await expect(host.get(Info, {}, context)).rejects.toMatchObject({
|
|
code: "not_ready",
|
|
message: `app.info: ${fix}`,
|
|
});
|
|
expect(() => host.hold(Info, {}, context)).toThrow(`app.info: ${fix}`);
|
|
await expect(host.registry(Names, context)).rejects.toMatchObject({
|
|
code: "not_ready",
|
|
message: `registry app.names: ${fix}`,
|
|
});
|
|
await expect(host.root((ctx) => ctx.get(Info, {}), context)).rejects.toMatchObject({ code: "not_ready" });
|
|
// A client with nothing to load calls `load([])`: then its requests wait as any does.
|
|
await host.load([], context);
|
|
expect(host.hold(Info, {}, context).state.value).toMatchObject({ kind: "failed", cause: { kind: "waiting" } });
|
|
await host.close(context);
|
|
});
|
|
});
|
|
|
|
describe("the start barrier", () => {
|
|
fakeTimers();
|
|
|
|
it("resolves once every host-lifetime object had its first outcome; a failed one is reported and retried", async () => {
|
|
let attempts = 0;
|
|
const { host, errors } = await loaded((h) => {
|
|
h.serve(Info, () => ({ title: constant("t") }));
|
|
h.serve(Counter, (_, self) => {
|
|
const attempt = ++attempts;
|
|
self.open(
|
|
async () => {
|
|
if (attempt === 1) throw new Error("offline");
|
|
return null;
|
|
},
|
|
() => {},
|
|
);
|
|
return { count: constant(attempt), bump: async () => null };
|
|
});
|
|
});
|
|
expect(failures(errors)).toEqual(["app.counter: transient"]);
|
|
const counter = host.get(Counter, {}, context);
|
|
await vi.advanceTimersByTimeAsync(1000);
|
|
expect((await counter).count.value).toBe(2);
|
|
});
|
|
|
|
it("is not held by a hanging open: its attempt is abandoned at readyTimeout", async () => {
|
|
const host = createHost({ ...noErrors, readyTimeout: 100 });
|
|
const facet = defineFacet((h) =>
|
|
h.serve(Info, (_, self) => {
|
|
self.open(
|
|
() => new Promise(() => {}),
|
|
() => {},
|
|
);
|
|
return { title: constant("t") };
|
|
}),
|
|
);
|
|
let done = false;
|
|
void host.load([{ name: "a", facet }], context).then(() => {
|
|
done = true;
|
|
});
|
|
await vi.advanceTimersByTimeAsync(99);
|
|
expect(done).toBe(false);
|
|
await vi.advanceTimersByTimeAsync(1);
|
|
expect(done).toBe(true);
|
|
});
|
|
|
|
it("waits for a local blocker's first outcome: a host object parked on another one here (CodeMode → Harness)", async () => {
|
|
const Harness = local("app.harness", { version: 1, params: {}, members: { n: int() } });
|
|
const CodeMode = local("app.codemode", { version: 1, params: {}, members: { n: int() } });
|
|
const opened = Promise.withResolvers<void>();
|
|
const host = createHost(noErrors);
|
|
const facet = defineFacet((h) => {
|
|
h.serve(Harness, (_, self) => {
|
|
self.open(
|
|
() => opened.promise,
|
|
() => {},
|
|
);
|
|
return { n: constant(1) };
|
|
});
|
|
h.serve(CodeMode, (_, self) => ({ n: self.get(Harness, {}).n }));
|
|
});
|
|
let done = false;
|
|
const loading = host.load([{ name: "@app/core", facet }], context).then(() => {
|
|
done = true;
|
|
});
|
|
await flushed();
|
|
expect(done).toBe(false);
|
|
opened.resolve();
|
|
await loading;
|
|
expect(host.hold(CodeMode, {}, context).state.value).toEqual({ kind: "live" });
|
|
});
|
|
|
|
it("ends its pass at a local blocker's own failure: the dependent parked failed { waiting } has its first outcome", async () => {
|
|
const Harness = local("app.harness", { version: 1, params: {}, members: { n: int() } });
|
|
const CodeMode = local("app.codemode", { version: 1, params: {}, members: { n: int() } });
|
|
const { host, errors } = await loaded((h) => {
|
|
h.serve(Harness, (_, self) => {
|
|
self.open(
|
|
async () => {
|
|
throw new Error("offline");
|
|
},
|
|
() => {},
|
|
);
|
|
return { n: constant(1) };
|
|
});
|
|
h.serve(CodeMode, (_, self) => ({ n: self.get(Harness, {}).n }));
|
|
});
|
|
expect(failures(errors)).toEqual(["app.harness: transient"]);
|
|
expect(host.hold(CodeMode, {}, context).state.value).toMatchObject({
|
|
kind: "failed",
|
|
cause: { kind: "waiting", error: { message: "waiting for app.harness: app.harness: offline" } },
|
|
});
|
|
});
|
|
|
|
it("ends its pass at a remote blocker: a wait on another host is a first outcome", async () => {
|
|
const CodeMode = local("app.codemode", { version: 1, params: {}, members: { n: int() } });
|
|
const host = createHost<TestPrincipal>(noErrors);
|
|
const [a, b] = memoryLinkPair();
|
|
host.connect(a, { principal: { kind: "server" }, codec: testCodec() }, context);
|
|
const parent = new RawPeer(b);
|
|
const facet = defineFacet<TestPrincipal>((h) =>
|
|
h.serve(CodeMode, (_, self) => ({ n: self.get(Notes, { session: "S1" }).count })),
|
|
);
|
|
await host.load([{ name: "@app/codemode", facet }], context);
|
|
expect(await parent.next("hold")).toMatchObject({ type: Notes.name });
|
|
expect(host.hold(CodeMode, {}, context).state.value).toEqual({ kind: "loading" });
|
|
});
|
|
|
|
it("is settled by host.close: an object of its cohort that ends before its first outcome settles it as ended", async () => {
|
|
const host = createHost(noErrors);
|
|
const facet = defineFacet((h) =>
|
|
h.serve(Info, (_, self) => {
|
|
self.open(
|
|
() => new Promise(() => {}),
|
|
() => {},
|
|
);
|
|
return { title: constant("t") };
|
|
}),
|
|
);
|
|
let done = false;
|
|
const loading = host.load([{ name: "a", facet }], context).then(() => {
|
|
done = true;
|
|
});
|
|
await flushed();
|
|
expect(done).toBe(false);
|
|
await host.close(context);
|
|
await loading;
|
|
expect(done).toBe(true);
|
|
});
|
|
|
|
it("waits for a local blocker parked failed { waiting } on one still in its first attempt (fix round 1, Sol 4)", async () => {
|
|
const Core = local("app.core", { version: 1, params: {}, members: { n: int() } });
|
|
const Middle = local("app.middle", { version: 1, params: { id: int() }, members: { n: int() } });
|
|
const Slow = local("app.slow", { version: 1, params: { id: int() }, members: { n: int() } });
|
|
const Missing = local("app.missing", { version: 1, params: {}, members: { n: int() } });
|
|
const gate = Promise.withResolvers<void>();
|
|
const host = createHost(noErrors);
|
|
const facet = defineFacet((h) => {
|
|
h.serve(Slow, (_, self) => {
|
|
self.open(
|
|
() => gate.promise,
|
|
() => {},
|
|
);
|
|
return { n: constant(1) };
|
|
});
|
|
h.serve(Middle, ({ id }, self) => {
|
|
const slow = self.get(Slow, { id });
|
|
self.get(Missing, {});
|
|
return { n: slow.n };
|
|
});
|
|
h.serve(Core, (_, self) => ({ n: self.get(Middle, { id: 1 }).n }));
|
|
});
|
|
let done = false;
|
|
const loading = host.load([{ name: "@app/core", facet }], context).then(() => {
|
|
done = true;
|
|
});
|
|
await flushed();
|
|
expect(host.hold(Middle, { id: 1 }, context).state.value).toMatchObject({
|
|
kind: "failed",
|
|
cause: { kind: "waiting" },
|
|
});
|
|
expect(done).toBe(false);
|
|
gate.resolve();
|
|
await loading;
|
|
await host.close(context);
|
|
});
|
|
|
|
it("waits for it whichever it declares first (fix round 1, Astra 3)", async () => {
|
|
const A = local("app.a", { version: 1, params: {}, members: {} });
|
|
const B = local("app.b", { version: 1, params: { id: int() }, members: {} });
|
|
const Slow = local("app.slow", { version: 1, params: { id: int() }, members: {} });
|
|
const Missing = local("app.missing", { version: 1, params: {}, members: {} });
|
|
const gate = Promise.withResolvers<void>();
|
|
const host = createHost(noErrors);
|
|
const facet = defineFacet((h) => {
|
|
h.serve(A, (_, self) => {
|
|
self.get(B, { id: 1 });
|
|
return {};
|
|
});
|
|
h.serve(B, (_, self) => {
|
|
self.get(Missing, {});
|
|
self.get(Slow, { id: 1 });
|
|
return {};
|
|
});
|
|
h.serve(Slow, (_, self) => {
|
|
self.open(
|
|
() => gate.promise,
|
|
() => {},
|
|
);
|
|
return {};
|
|
});
|
|
});
|
|
let done = false;
|
|
const loading = host.load([{ name: "@app/a", facet }], context).then(() => {
|
|
done = true;
|
|
});
|
|
await flushed();
|
|
const premature = done;
|
|
gate.resolve();
|
|
await loading;
|
|
await host.close(context);
|
|
expect(premature).toBe(false);
|
|
});
|
|
});
|
|
|
|
describe("startup across hosts (dependency model §6)", () => {
|
|
// Timers never advance: nothing here may wait for a deadline.
|
|
fakeTimers();
|
|
|
|
/** On the server, waiting for a worker's `Agent`. */
|
|
const Bootstrap = local("app.bootstrap", { version: 1, params: {}, members: { n: int() } });
|
|
/** On worker S1, waiting for the server's `Catalog`. */
|
|
const Agent = remote("app.agent", {
|
|
version: 1,
|
|
params: { session: str() },
|
|
hostAddress: ({ session }) => Session({ id: session }),
|
|
variants: { v: { state: { n: int() } } },
|
|
});
|
|
const Catalog = remote("app.catalog", {
|
|
version: 1,
|
|
params: {},
|
|
hostAddress: () => Server(),
|
|
variants: { v: { state: { n: int() } } },
|
|
});
|
|
|
|
it("starts two hosts that would have deadlocked (Sol's probe): an acyclic graph needs no timeout", async () => {
|
|
const { server, errors } = await relayed({
|
|
server: (h) => {
|
|
h.serve(Bootstrap, (_, self) => ({ n: self.get(Agent, { session: "S1" }).n }));
|
|
h.serve(Catalog, () => ({ variant: "v", members: { n: constant(7) } }));
|
|
},
|
|
worker: (h) => {
|
|
// The worker's own host object waits for the server's catalog during the worker's load.
|
|
h.serve(Agent, (_params, _principal, self) => ({ variant: "v", members: { n: self.get(Catalog, {}).n } }));
|
|
},
|
|
});
|
|
await transported();
|
|
await transported();
|
|
const bootstrap = server.hold(Bootstrap, {}, context);
|
|
expect(bootstrap.state.value).toEqual({ kind: "live" });
|
|
expect(bootstrap.n.value).toBe(7);
|
|
expect([errors.server, errors.worker]).toEqual([[], []]);
|
|
});
|
|
|
|
it("places a hold arriving before the publication here, not over the parent", async () => {
|
|
const worker = createHost<TestPrincipal>({
|
|
address: Session({ id: "S1" }),
|
|
principal: { kind: "worker", session: "S1" },
|
|
...noErrors,
|
|
});
|
|
const [a, b] = memoryLinkPair();
|
|
worker.connect(a, { principal: { kind: "server" }, codec: testCodec() }, context);
|
|
const parent = new RawPeer(b);
|
|
const [c, d] = memoryLinkPair();
|
|
worker.accept(c, { principal: { kind: "owner" }, codec: testCodec() }, context);
|
|
const child = new RawPeer(d);
|
|
child.send({
|
|
kind: "hold",
|
|
hold: 0,
|
|
type: PlacedNotes.name,
|
|
version: PlacedNotes.version,
|
|
fingerprint: PlacedNotes.fingerprint,
|
|
params: jsonPart({ session: "S1" }),
|
|
address: { kind: "session", params: { id: "S1" } },
|
|
});
|
|
await transported();
|
|
await worker.load(
|
|
[{ name: "@test/w", facet: defineFacet<TestPrincipal>((h) => serveNotes(h, PlacedNotes)) }],
|
|
context,
|
|
);
|
|
expect((await child.next("bound")).holds[0]).toMatchObject({ hold: 0 });
|
|
expect(parent.kinds()).toEqual(["hello"]);
|
|
});
|
|
|
|
it("wakes a parked chain over three hosts in order, its causes passed on (tab → worker → server)", async () => {
|
|
const Chat = remote("app.chat", {
|
|
version: 1,
|
|
params: { session: str() },
|
|
hostAddress: ({ session }) => Session({ id: session }),
|
|
variants: { v: { state: { n: int() } } },
|
|
});
|
|
const order: string[] = [];
|
|
let catalogs = 0;
|
|
const { tab } = await relayed({
|
|
server: (h) =>
|
|
h.serve(Catalog, (_params, _principal, self) => {
|
|
const attempt = ++catalogs;
|
|
self.open(
|
|
async () => {
|
|
if (attempt === 1) throw new Error("offline");
|
|
},
|
|
() => {},
|
|
);
|
|
self.onLive(async () => void order.push("catalog"));
|
|
return { variant: "v", members: { n: constant(attempt) } };
|
|
}),
|
|
worker: (h) =>
|
|
h.serve(Chat, (_params, _principal, self) => {
|
|
const catalog = self.get(Catalog, {});
|
|
self.onLive(async () => void order.push("chat"));
|
|
return { variant: "v", members: { n: catalog.n } };
|
|
}),
|
|
});
|
|
const view = tab.hold(Chat, { session: "S1" }, context);
|
|
for (let round = 0; round < 5; round++) await transported();
|
|
expect(view.state.value).toMatchObject({
|
|
kind: "failed",
|
|
cause: { kind: "waiting", error: { message: "waiting for app.catalog: app.catalog: offline" } },
|
|
retryAt: undefined,
|
|
});
|
|
// The catalog's own backoff (1 s) is the only timer: it retries, and the chain goes on in order.
|
|
await vi.advanceTimersByTimeAsync(1_000);
|
|
for (let round = 0; round < 5; round++) await transported();
|
|
expect(view.state.value).toEqual({ kind: "live" });
|
|
expect(order).toEqual(["catalog", "chat"]);
|
|
expect(view.n.value).toBe(2);
|
|
});
|
|
});
|
|
|
|
describe("get", () => {
|
|
it("waits for an unserved type and a contract mismatch until its readyTimeout, then rejects with the cause", async () => {
|
|
const { host } = await loaded((h) => h.serve(Info, () => ({ title: constant("t") })));
|
|
const quick = { readyTimeout: 5 };
|
|
expect(await rejection(host.get(Page, { id: 1 }, context, quick))).toEqual({
|
|
code: "unavailable",
|
|
message: "app.page is not served here",
|
|
});
|
|
const InfoV2 = local("app.info", { version: 2, params: {}, members: { title: str() } });
|
|
const Changed = local("app.info", { version: 1, params: {}, members: { title: int() } });
|
|
expect(await rejection(host.get(InfoV2, {}, context, quick))).toMatchObject({ code: "contract_mismatch" });
|
|
expect(await rejection(host.get(Changed, {}, context, quick))).toMatchObject({ code: "contract_mismatch" });
|
|
const Same = local("app.info", { version: 1, params: {}, members: { title: str() } });
|
|
expect((await host.get(Same, {}, context)).title.value).toBe("t");
|
|
});
|
|
|
|
it("checks params with L1 (invalid/params)", async () => {
|
|
const { host } = await loaded((h) => h.serve(Page, ({ id }) => ({ id: constant(id) })));
|
|
// @ts-expect-error: a string is not an int
|
|
expect(await rejection(host.get(Page, { id: "x" }, context))).toMatchObject({
|
|
code: "invalid",
|
|
reason: "params",
|
|
});
|
|
});
|
|
});
|
|
|
|
describe("deadlines of gets", () => {
|
|
fakeTimers();
|
|
|
|
it("reject with unavailable while no attempt has failed (the last failure: see attempt.test.ts)", async () => {
|
|
const gate = Promise.withResolvers<void>();
|
|
const { host } = await loaded((h) => serveGated(h, gate.promise, []));
|
|
const waiting = rejection(host.get(Svc, { id: 1 }, context, { readyTimeout: 500 }));
|
|
await vi.advanceTimersByTimeAsync(500);
|
|
expect(await waiting).toEqual({ code: "unavailable", message: "app.svc(1): did not become live within 500 ms" });
|
|
});
|
|
|
|
it("are per getter: the first to pass releases only its own reference, the attempt goes on for the other", async () => {
|
|
const gate = Promise.withResolvers<void>();
|
|
const closes: string[] = [];
|
|
const { host } = await loaded((h) => serveGated(h, gate.promise, closes), { readyTimeout: 10_000 });
|
|
const early = rejection(host.get(Svc, { id: 1 }, context, { readyTimeout: 100 }));
|
|
const late = host.get(Svc, { id: 1 }, context, { readyTimeout: 1000 });
|
|
await vi.advanceTimersByTimeAsync(100);
|
|
expect(await early).toMatchObject({ code: "unavailable" });
|
|
gate.resolve();
|
|
expect((await late).n.value).toBe(1);
|
|
expect(closes).toEqual([]);
|
|
});
|
|
});
|
|
|
|
describe("close", () => {
|
|
it("ends every object, the last to go live first, and rejects new calls", async () => {
|
|
const Outer = local("app.outer", { version: 1, params: {}, members: { title: str() } });
|
|
const { host } = await loaded((h) => {
|
|
h.serve(Info, () => ({ title: constant("info") }));
|
|
h.serve(Outer, (_, self) => ({ title: self.get(Info, {}).title }));
|
|
});
|
|
const ends: string[] = [];
|
|
const outer = await host.get(Outer, {}, context);
|
|
const info = await host.get(Info, {}, context);
|
|
void (await outer.title.watch(context)).closed.then(() => ends.push("outer"));
|
|
void (await info.title.watch(context)).closed.then(() => ends.push("info"));
|
|
await host.close(context);
|
|
expect(ends).toEqual(["outer", "info"]);
|
|
expect(await rejection(host.get(Info, {}, context))).toEqual({
|
|
code: "released",
|
|
reason: "closed",
|
|
message: "the host is closed",
|
|
});
|
|
expect(() => info.title.value).toThrow("app.info: ended");
|
|
});
|
|
|
|
it("ends every object at once: calls fail released before any close ran", async () => {
|
|
const closing = Promise.withResolvers<void>();
|
|
const { host } = await loaded((h) =>
|
|
h.serve(Counter, (_, self) => {
|
|
self.open(
|
|
async () => null,
|
|
() => closing.promise,
|
|
);
|
|
return { count: constant(1), bump: async () => null };
|
|
}),
|
|
);
|
|
const counter = await host.get(Counter, {}, context);
|
|
const closed = host.close(context);
|
|
expect(await rejection(counter.bump(context))).toEqual({
|
|
code: "released",
|
|
reason: "ended",
|
|
message: "app.counter: ended",
|
|
});
|
|
closing.resolve();
|
|
await closed;
|
|
});
|
|
|
|
it("aborts a pending open and does not wait for it; its late value is closed with close", async () => {
|
|
const gate = Promise.withResolvers<void>();
|
|
const closes: string[] = [];
|
|
const { host } = await loaded((h) => serveGated(h, gate.promise, closes));
|
|
const got = rejection(host.get(Svc, { id: 1 }, context));
|
|
await flushed();
|
|
await host.close(context);
|
|
expect(await got).toEqual({ code: "released", reason: "ended", message: "app.svc(1): ended" });
|
|
gate.resolve();
|
|
await flushed();
|
|
expect(closes).toEqual(["svc 1 close"]);
|
|
});
|
|
|
|
it("joins an ending running in the background", async () => {
|
|
const closing = Promise.withResolvers<void>();
|
|
const { host } = await loaded((h) =>
|
|
h.serve(Svc, ({ id }, self) => {
|
|
self.open(
|
|
async () => null,
|
|
() => closing.promise,
|
|
);
|
|
return { n: constant(id) };
|
|
}),
|
|
);
|
|
(await host.get(Svc, { id: 1 }, context)).release();
|
|
let closed = false;
|
|
const close = host.close(context).then(() => {
|
|
closed = true;
|
|
});
|
|
await flushed();
|
|
expect(closed).toBe(false);
|
|
closing.resolve();
|
|
await close;
|
|
});
|
|
|
|
it("closes object by object, a dependent before its dependencies", async () => {
|
|
const View = local("app.view", { version: 1, params: { id: int() }, members: { n: int() } });
|
|
const events: string[] = [];
|
|
/** A native whose close records when it starts and, a turn later, when it ends. */
|
|
const open = (self: Self, name: string): void => {
|
|
self.open(
|
|
async () => null,
|
|
async () => {
|
|
events.push(`${name} closing`);
|
|
await flushed();
|
|
events.push(`${name} closed`);
|
|
},
|
|
);
|
|
};
|
|
const { host } = await loaded((h) => {
|
|
h.serve(Svc, ({ id }, self) => {
|
|
open(self, `svc ${id}`);
|
|
return { n: constant(id) };
|
|
});
|
|
h.serve(View, ({ id }, self) => {
|
|
open(self, `view ${id}`);
|
|
return { n: self.get(Svc, { id }).n };
|
|
});
|
|
});
|
|
await host.get(View, { id: 1 }, context);
|
|
await host.get(View, { id: 2 }, context);
|
|
await host.close(context);
|
|
expect(events).toEqual([
|
|
"view 2 closing",
|
|
"view 2 closed",
|
|
"svc 2 closing",
|
|
"svc 2 closed",
|
|
"view 1 closing",
|
|
"view 1 closed",
|
|
"svc 1 closing",
|
|
"svc 1 closed",
|
|
]);
|
|
});
|
|
});
|
|
|
|
describe("close, its order and its races", () => {
|
|
it("returns one closing to every caller: a second close waits for the first", async () => {
|
|
const closing = Promise.withResolvers<void>();
|
|
const { host } = await loaded((h) =>
|
|
h.serve(Counter, (_, self) => {
|
|
self.open(
|
|
async () => null,
|
|
() => closing.promise,
|
|
);
|
|
return { count: constant(1), bump: async () => null };
|
|
}),
|
|
);
|
|
const first = host.close(context);
|
|
let second = false;
|
|
void host.close(context).then(() => {
|
|
second = true;
|
|
});
|
|
await flushed();
|
|
expect(second).toBe(false);
|
|
closing.resolve();
|
|
await first;
|
|
await flushed();
|
|
expect(second).toBe(true);
|
|
});
|
|
|
|
it("tears down what a failed dependent holds with close, and waits for it", async () => {
|
|
const Pair = local("app.pair", { version: 1, params: { id: int() }, members: { n: int() } });
|
|
const closing = Promise.withResolvers<void>();
|
|
const closes: string[] = [];
|
|
const { host } = await loaded((h) => {
|
|
h.serve(Svc, ({ id }, self) => {
|
|
self.open(
|
|
async () => null,
|
|
async (_value, reason) => {
|
|
closes.push(`svc ${reason.kind}`);
|
|
await closing.promise;
|
|
},
|
|
);
|
|
return { n: constant(id) };
|
|
});
|
|
h.serve(Core, (_, self) => {
|
|
self.open(
|
|
async () => {
|
|
throw new Error("offline");
|
|
},
|
|
() => {},
|
|
);
|
|
return { n: constant(0) };
|
|
});
|
|
h.serve(Pair, ({ id }, self) => ({
|
|
n: self.derive([self.get(Svc, { id }).n, self.get(Core, {}).n], ([n]) => n),
|
|
}));
|
|
});
|
|
// The pair fails on the core (R18) and keeps its references: svc(1), live before the pair, held by it alone.
|
|
const svc = await host.get(Svc, { id: 1 }, context);
|
|
void host.get(Pair, { id: 1 }, context).catch(() => {});
|
|
await flushed();
|
|
svc.release();
|
|
let closed = false;
|
|
const close = host.close(context).then(() => {
|
|
closed = true;
|
|
});
|
|
await flushed();
|
|
expect([closes, closed]).toEqual([["svc close"], false]);
|
|
closing.resolve();
|
|
await close;
|
|
});
|
|
|
|
it("goes on when onError throws: failures are still classified, teardowns still finish", async () => {
|
|
const host = createHost({
|
|
onError: () => {
|
|
throw new Error("onError threw");
|
|
},
|
|
});
|
|
const facet = defineFacet((h) =>
|
|
h.serve(Svc, ({ id }, self) => {
|
|
self.open(
|
|
async () => {
|
|
if (id === 2) throw broken(new Error("no config"));
|
|
return null;
|
|
},
|
|
() => {
|
|
throw new Error("close threw");
|
|
},
|
|
);
|
|
return { n: constant(id) };
|
|
}),
|
|
);
|
|
await host.load([{ name: "a", facet }], context);
|
|
// The background end reports its throwing close; the next instance waits for that teardown.
|
|
(await host.get(Svc, { id: 1 }, context)).release();
|
|
expect((await host.get(Svc, { id: 1 }, context)).n.value).toBe(1);
|
|
expect(await rejection(host.get(Svc, { id: 2 }, context))).toEqual({
|
|
code: "unavailable",
|
|
message: "app.svc(2): no config",
|
|
});
|
|
});
|
|
|
|
describe("with a backoff window running", () => {
|
|
fakeTimers();
|
|
|
|
it("ends an object waiting for its window and clears every timer", async () => {
|
|
const { host } = await loaded((h) =>
|
|
h.serve(Svc, (_, self) => {
|
|
self.open(
|
|
async () => {
|
|
throw new Error("offline");
|
|
},
|
|
() => {},
|
|
);
|
|
return { n: constant(0) };
|
|
}),
|
|
);
|
|
const first = rejection(host.get(Svc, { id: 1 }, context, { readyTimeout: 100 }));
|
|
await vi.advanceTimersByTimeAsync(100);
|
|
await first;
|
|
const waiting = rejection(host.get(Svc, { id: 1 }, context));
|
|
await flushed();
|
|
await host.close(context);
|
|
expect(await waiting).toEqual({ code: "released", reason: "ended", message: "app.svc(1): ended" });
|
|
expect(vi.getTimerCount()).toBe(0);
|
|
});
|
|
});
|
|
});
|
|
|
|
describe("getters", () => {
|
|
it("one cancelled while another waits for the same object: the other gets it, the object lives", async () => {
|
|
const { context: cancellable, cancel } = withCancel(context);
|
|
const factory = vi.fn((_: { id: number }) => {
|
|
cancel(new Error("cancelled"));
|
|
return { id: constant(1) };
|
|
});
|
|
const { host } = await loaded((h) => h.serve(Page, factory));
|
|
const [cancelled, kept] = await Promise.allSettled([
|
|
host.get(Page, { id: 1 }, cancellable),
|
|
host.get(Page, { id: 1 }, context),
|
|
]);
|
|
expect(cancelled).toMatchObject({ status: "rejected", reason: new Error("cancelled") });
|
|
expect(kept.status === "fulfilled" && kept.value.id.value).toBe(1);
|
|
await host.get(Page, { id: 1 }, context);
|
|
expect(factory).toHaveBeenCalledTimes(1);
|
|
});
|
|
});
|
|
|
|
forEachCodec(() => {
|
|
describe("principals and where a reference goes", () => {
|
|
const Who = remote("app.who", {
|
|
version: 1,
|
|
params: { id: int() },
|
|
hostAddress: () => Server(),
|
|
variants: { v: { state: { who: str() } } },
|
|
});
|
|
const Outer = remote("app.outer", {
|
|
version: 1,
|
|
params: {},
|
|
hostAddress: () => Server(),
|
|
variants: { v: { state: { who: str() } } },
|
|
});
|
|
const Inner = local("app.inner", { version: 1, params: { id: int() }, members: { who: str() } });
|
|
const Desk = local("app.desk", { 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()) } } },
|
|
});
|
|
type P = TestPrincipal;
|
|
|
|
/** Serves every type above: `Who` shows the principal its object is for. */
|
|
function serveAll(h: FacetHost<P>, factories: string[]) {
|
|
h.serve(Who, ({ id }, principal) => {
|
|
factories.push(`${id} ${principal.kind}`);
|
|
return { variant: "v", members: { who: constant(principal.kind) } };
|
|
});
|
|
h.serve(Outer, (_, _principal, self) => ({ variant: "v", members: { who: self.get(Who, { id: 3 }).who } }));
|
|
h.serve(Inner, ({ id }, self) => ({ who: self.get(Who, { id }).who }));
|
|
const who = async (ctx: Ctx<P>) => {
|
|
const ref = await ctx.get(Who, { id: 2 });
|
|
return ref.who.value;
|
|
};
|
|
h.serve(Desk, () => ({ who }));
|
|
h.serve(Gate, () => ({ variant: "v", members: { who } }));
|
|
}
|
|
|
|
it("host.get acts as the host's principal; a host without one (a client) serves no remote type", async () => {
|
|
const factories: string[] = [];
|
|
const { server } = await linked((h) => serveAll(h, factories));
|
|
expect((await server.get(Who, { id: 1 }, context)).who.value).toBe("server");
|
|
const client = createHost<P>({ onError: () => {} });
|
|
const report = await client.load(
|
|
[{ name: "@test/app", facet: defineFacet<P>((h) => serveAll(h, [])) }],
|
|
context,
|
|
);
|
|
expect(report.failed).toEqual([
|
|
{
|
|
plugin: "@test/app",
|
|
error: expect.objectContaining({
|
|
message: 'plugin "@test/app": app.who: a host without a principal serves only local types',
|
|
}),
|
|
},
|
|
]);
|
|
await client.close(context);
|
|
});
|
|
|
|
it("a host without an address serves no remote type, whatever its principal", async () => {
|
|
const host = createHost<P>({ principal: { kind: "server" }, onError: () => {} });
|
|
const report = await host.load(
|
|
[{ name: "@test/app", facet: defineFacet<P>((h) => serveAll(h, [])) }],
|
|
context,
|
|
);
|
|
expect(report.failed).toEqual([
|
|
{
|
|
plugin: "@test/app",
|
|
error: expect.objectContaining({
|
|
message: 'plugin "@test/app": app.who: a host without an address serves only local types',
|
|
}),
|
|
},
|
|
]);
|
|
await host.close(context);
|
|
});
|
|
|
|
it("ctx.get acts as ctx.principal: a guest's call gets the guest's view; a local handler's acts as the host", async () => {
|
|
const { server, client } = await linked((h) => serveAll(h, []), { child: { kind: "guest", session: "S1" } });
|
|
expect(await (await client.get(Gate, {}, context)).who(context)).toBe("guest");
|
|
expect(await (await server.get(Desk, {}, context)).who(context)).toBe("server");
|
|
});
|
|
|
|
it("self.get acts as a remote object's principal, a local object's as the host's", async () => {
|
|
const { server, client } = await linked((h) => serveAll(h, []), { child: { kind: "guest", session: "S1" } });
|
|
expect((await client.get(Outer, {}, context)).who.value).toBe("guest");
|
|
expect((await server.get(Inner, { id: 4 }, context)).who.value).toBe("server");
|
|
});
|
|
|
|
it("identity includes the principal key: equal keys share one object", async () => {
|
|
const factories: string[] = [];
|
|
const host = createHost<P>({
|
|
address: Server(),
|
|
principal: { kind: "guest", session: "S1" },
|
|
principalKey: (principal) => principal.kind,
|
|
onError: () => {},
|
|
});
|
|
await host.load([{ name: "@test/app", facet: defineFacet<P>((h) => serveAll(h, factories)) }], context);
|
|
const [a, b] = memoryLinkPair();
|
|
host.accept(a, { principal: { kind: "guest", session: "S2" }, codec: testCodec() }, context);
|
|
const peer = new RawPeer(b);
|
|
await host.get(Who, { id: 1 }, context);
|
|
peer.send({
|
|
kind: "hold",
|
|
hold: 0,
|
|
type: Who.name,
|
|
version: Who.version,
|
|
fingerprint: Who.fingerprint,
|
|
params: jsonPart({ id: 1 }),
|
|
address: SERVER_ADDRESS,
|
|
});
|
|
await peer.next("bound");
|
|
expect(factories).toEqual(["1 guest"]);
|
|
const [c, d] = memoryLinkPair();
|
|
host.accept(c, { principal: { kind: "owner" }, codec: testCodec() }, context);
|
|
const owner = new RawPeer(d);
|
|
owner.send({
|
|
kind: "hold",
|
|
hold: 0,
|
|
type: Who.name,
|
|
version: Who.version,
|
|
fingerprint: Who.fingerprint,
|
|
params: jsonPart({ id: 1 }),
|
|
address: SERVER_ADDRESS,
|
|
});
|
|
await owner.next("bound");
|
|
expect(factories).toEqual(["1 guest", "1 owner"]);
|
|
});
|
|
|
|
it("remote objects with empty params are not activated by load; they live while referenced", async () => {
|
|
const factories: string[] = [];
|
|
const Empty = remote("app.empty", {
|
|
version: 1,
|
|
params: {},
|
|
hostAddress: () => Server(),
|
|
variants: { v: { state: { n: int() } } },
|
|
});
|
|
const { server } = await linked((h) =>
|
|
h.serve(Empty, (_, principal) => {
|
|
factories.push(principal.kind);
|
|
return { variant: "v", members: { n: constant(1) } };
|
|
}),
|
|
);
|
|
expect(factories).toEqual([]);
|
|
(await server.get(Empty, {}, context)).release();
|
|
await flushed();
|
|
await server.get(Empty, {}, context);
|
|
expect(factories).toEqual(["server", "server"]);
|
|
});
|
|
|
|
it("a local type never crosses a link; a remote type without a parent link waits for one", async () => {
|
|
const { client } = await linked(() => {});
|
|
expect(client.hold(Inner, { id: 1 }, context).state.value).toMatchObject({
|
|
kind: "failed",
|
|
cause: { kind: "waiting" },
|
|
});
|
|
expect(await rejection(client.get(Inner, { id: 1 }, context, { readyTimeout: 5 }))).toEqual({
|
|
code: "unavailable",
|
|
message: "app.inner is not served here",
|
|
});
|
|
const alone = createHost<P>({ onError: () => {} });
|
|
await alone.load([], context);
|
|
expect(alone.hold(Who, { id: 1 }, context).state.value).toEqual({ kind: "loading" });
|
|
expect(await rejection(alone.get(Who, { id: 1 }, context, { readyTimeout: 10 }))).toEqual({
|
|
code: "unavailable",
|
|
message: "app.who(1): did not become live within 10 ms",
|
|
});
|
|
});
|
|
|
|
it("connect takes one open parent link", async () => {
|
|
const { client } = await linked(() => {});
|
|
expect(() =>
|
|
client.connect(memoryLinkPair()[0], { principal: { kind: "server" }, codec: testCodec() }, context),
|
|
).toThrow("connect: this host has an open parent link");
|
|
});
|
|
|
|
it("close closes the links first: the consumer sees the link drop, no gone", async () => {
|
|
const { server, client } = await linked((h) => serveAll(h, []));
|
|
const who = await client.get(Who, { id: 1 }, context);
|
|
const watch = await who.who.watch(context);
|
|
await server.close(context);
|
|
await transported();
|
|
expect(who.state.value).toEqual({ kind: "loading" });
|
|
expect(who.who.value).toBe("owner");
|
|
expect(await Promise.race([watch.closed, transported().then(() => "open")])).toBe("open");
|
|
});
|
|
});
|
|
});
|
|
|
|
describe("host.registry and the host's close (fix round 1)", () => {
|
|
const Absent = defineRegistry<number, number>("app.absent", { version: 1 });
|
|
|
|
it("ends a waiting host.registry at the close with released/closed, its deadline timer with it (Sol 5, Astra 5)", async () => {
|
|
const host = createHost(noErrors);
|
|
await host.load([], context);
|
|
let outcome: unknown;
|
|
void rejection(host.registry(Absent, context, { readyTimeout: 60_000 })).then((error) => {
|
|
outcome = error;
|
|
});
|
|
await flushed();
|
|
await host.close(context);
|
|
await flushed();
|
|
expect(outcome).toMatchObject({ code: "released", reason: "closed" });
|
|
});
|
|
});
|
|
|
|
describe("the selected provider's own checks for a request made here (fix round 1)", () => {
|
|
const Here = hostKind("app-here");
|
|
const Elsewhere = hostKind("app-elsewhere");
|
|
const shape = { version: 1, params: {}, variants: { view: { state: { n: int() } } } } as const;
|
|
|
|
it("waits a same-host remote request the provider's code places elsewhere: invalid/host_address (Astra 4)", async () => {
|
|
const Consumer = remote("app.placed", { ...shape, hostAddress: () => Here() });
|
|
const Provider = remote("app.placed", { ...shape, hostAddress: () => Elsewhere() });
|
|
const host = createHost({ address: Here(), principal: { kind: "owner" }, onError: () => {} });
|
|
const facet = defineFacet((h) => h.serve(Provider, () => ({ variant: "view", members: { n: constant(1) } })));
|
|
await host.load([{ name: "@app/provider", facet }], context);
|
|
const ref = host.hold(Consumer, {}, context);
|
|
await flushed();
|
|
expect(ref.state.value).toMatchObject({
|
|
kind: "failed",
|
|
cause: { kind: "waiting", error: { code: "invalid", reason: "host_address" } },
|
|
});
|
|
ref.release();
|
|
await host.close(context);
|
|
});
|
|
|
|
it("fails broken when the provider's hostAddress throws: its own code (Sol 7)", async () => {
|
|
const Consumer = remote("app.remote", { ...shape, hostAddress: () => Server() });
|
|
const Provider = remote("app.remote", {
|
|
...shape,
|
|
hostAddress: () => {
|
|
throw new Error("provider address bug");
|
|
},
|
|
});
|
|
const server = createHost({ address: Server(), principal: { kind: "server" }, onError: () => {} });
|
|
const client = createHost({ onError: () => {} });
|
|
const facet = defineFacet((h) => h.serve(Provider, () => ({ variant: "view", members: { n: constant(1) } })));
|
|
await server.load([{ name: "@app/provider", facet }], context);
|
|
await client.load([], context);
|
|
const [down, up] = memoryLinkPair();
|
|
server.accept(down, { principal: { kind: "server" } }, context);
|
|
client.connect(up, { principal: { kind: "server" } }, context);
|
|
const held = client.hold(Consumer, {}, context);
|
|
await transported();
|
|
expect(held.state.value).toMatchObject({
|
|
kind: "failed",
|
|
cause: { kind: "broken", error: { message: "app.remote: provider address bug" } },
|
|
});
|
|
// The same throw for a request made on the provider's host fails it broken too.
|
|
const here = server.hold(Consumer, {}, context);
|
|
expect(here.state.value).toMatchObject({ kind: "failed", cause: { kind: "broken" } });
|
|
await client.close(context);
|
|
await server.close(context);
|
|
});
|
|
});
|
|
|
|
describe('a load whose setup throws (decision "Per plugin at startup and reload")', () => {
|
|
it("leaves that plugin out and reports it; the others publish, the host and its links stay open", async () => {
|
|
const host = createHost<TestPrincipal>({ principal: { kind: "server" }, onError: () => {} });
|
|
const [a, b] = memoryLinkPair();
|
|
const [c, d] = memoryLinkPair();
|
|
host.accept(a, { principal: { kind: "owner" }, codec: testCodec() }, context);
|
|
host.connect(c, { principal: { kind: "server" }, codec: testCodec() }, context);
|
|
const child = new RawPeer(b);
|
|
const parent = new RawPeer(d);
|
|
const broken = defineFacet<TestPrincipal>(() => {
|
|
throw new Error("setup threw");
|
|
});
|
|
const report = await host.load(
|
|
[
|
|
{ name: "@test/broken", facet: broken },
|
|
{
|
|
name: "@test/app",
|
|
facet: defineFacet<TestPrincipal>((h) => h.serve(Info, () => ({ title: constant("t") }))),
|
|
},
|
|
],
|
|
context,
|
|
);
|
|
expect(report.failed).toEqual([
|
|
{ plugin: "@test/broken", error: expect.objectContaining({ message: 'plugin "@test/broken": setup threw' }) },
|
|
]);
|
|
await transported();
|
|
expect([child.closed, parent.closed]).toEqual([false, false]);
|
|
expect((await host.get(Info, {}, context)).title.value).toBe("t");
|
|
// What the left-out plugin would have served waits as any missing type does.
|
|
expect(host.hold(Page, { id: 1 }, context).state.value).toMatchObject({
|
|
kind: "failed",
|
|
cause: { kind: "waiting" },
|
|
});
|
|
await host.close(context);
|
|
});
|
|
});
|
|
|
|
describe("acquiring: hold and get", () => {
|
|
fakeTimers();
|
|
|
|
/** Serves `Svc`, whose open throws `offline` (broken when `failing.broken`) while `failing.left` is above 0. */
|
|
function serveFlaky(h: FacetHost, failing: { left: number; broken: boolean }): void {
|
|
h.serve(Svc, ({ id }, self) => {
|
|
self.open(
|
|
async () => {
|
|
if (failing.left-- <= 0) return;
|
|
throw failing.broken ? broken(new Error("offline")) : new Error("offline");
|
|
},
|
|
() => {},
|
|
);
|
|
return { n: constant(id) };
|
|
});
|
|
}
|
|
|
|
const offline = { code: "unavailable", message: "app.svc(1): offline" };
|
|
|
|
it("invalid params at once; a local type not served here: hold waits failed, get rejects at its readyTimeout", async () => {
|
|
const host = createHost<TestPrincipal>({ address: Server(), ...noErrors, principal: { kind: "server" } });
|
|
await host.load([{ name: "@test/app", facet: defineFacet<TestPrincipal>((h) => void serveNotes(h)) }], context);
|
|
const params = { code: "invalid", reason: "params" };
|
|
expect(() => host.hold(Notes, { session: 1 } as never, context)).toThrow("params");
|
|
expect(await rejection(host.get(Notes, { session: 1 } as never, context))).toMatchObject(params);
|
|
const unserved = { code: "unavailable", message: "app.info is not served here" };
|
|
expect(host.hold(Info, {}, context).state.value).toMatchObject({
|
|
kind: "failed",
|
|
cause: { kind: "waiting", error: unserved },
|
|
retryAt: undefined,
|
|
});
|
|
const got = rejection(host.get(Info, {}, context, { readyTimeout: 5 }));
|
|
await vi.advanceTimersByTimeAsync(5);
|
|
expect(await got).toEqual(unserved);
|
|
});
|
|
|
|
it("through transient failures: hold is failed with retryAt, then loading; get waits, then rejects with the last cause", async () => {
|
|
const { host } = await loaded((h) => serveFlaky(h, { left: 5, broken: false }));
|
|
const held = host.hold(Svc, { id: 1 }, context);
|
|
const got = rejection(host.get(Svc, { id: 1 }, context, { readyTimeout: 1_500 }));
|
|
await flushed();
|
|
expect(held.state.value).toMatchObject({
|
|
kind: "failed",
|
|
cause: { kind: "transient", error: offline },
|
|
retryAt: 1_000,
|
|
});
|
|
await vi.advanceTimersByTimeAsync(1_000);
|
|
await flushed();
|
|
expect(held.state.value).toMatchObject({ kind: "failed", retryAt: 3_000 });
|
|
await vi.advanceTimersByTimeAsync(500);
|
|
expect(await got).toEqual(offline);
|
|
// The getter released only its own reference: the hold follows the object's retries.
|
|
await vi.advanceTimersByTimeAsync(1_500);
|
|
expect(held.state.value).toMatchObject({ kind: "failed", retryAt: 7_000 });
|
|
});
|
|
|
|
it("a broken failure: hold is failed without retryAt; get releases its reference and rejects at once", async () => {
|
|
const { host } = await loaded((h) => serveFlaky(h, { left: 1, broken: true }));
|
|
const held = host.hold(Svc, { id: 1 }, context);
|
|
expect(await rejection(host.get(Svc, { id: 1 }, context))).toEqual(offline);
|
|
expect(held.state.value).toMatchObject({ kind: "failed", cause: { kind: "broken" }, retryAt: undefined });
|
|
});
|
|
|
|
it("a cancelled context: no effect on a hold; a get releases its reference and rejects", async () => {
|
|
const gate = Promise.withResolvers<void>();
|
|
const closes: string[] = [];
|
|
const { host } = await loaded((h) => serveGated(h, gate.promise, closes));
|
|
const { context: cancellable, cancel } = withCancel(context);
|
|
const held = host.hold(Svc, { id: 1 }, cancellable);
|
|
const got = host.get(Svc, { id: 2 }, cancellable);
|
|
cancel(new Error("user cancelled"));
|
|
await expect(got).rejects.toThrow("user cancelled");
|
|
gate.resolve();
|
|
await flushed();
|
|
expect(held.state.value).toEqual({ kind: "live" });
|
|
// The getter's object ended with its only reference: its late value was closed.
|
|
expect(closes).toEqual(["svc 2 end"]);
|
|
});
|
|
|
|
it("host.close: queued calls reject released, references go released, their watches end (race 10)", async () => {
|
|
const gate = Promise.withResolvers<void>();
|
|
const { host } = await loaded((h) => serveGated(h, gate.promise, []));
|
|
const local = host.hold(Svc, { id: 1 }, context);
|
|
const remoteHost = createHost<TestPrincipal>(noErrors);
|
|
await remoteHost.load([], context);
|
|
const remote = remoteHost.hold(Notes, { session: "S1" }, context);
|
|
const call = remote.wait(context);
|
|
const watches = [await local.state.watch(context), await remote.state.watch(context)];
|
|
await Promise.all([host.close(context), remoteHost.close(context)]);
|
|
expect(await rejection(call)).toEqual({ code: "released", reason: "closed", message: "the host is closed" });
|
|
expect([local.state.value, remote.state.value]).toMatchObject([{ kind: "ended" }, { kind: "ended" }]);
|
|
expect(await Promise.all(watches.map(({ closed }) => closed))).toEqual([{ kind: "ended" }, { kind: "ended" }]);
|
|
});
|
|
});
|
|
|
|
forEachCodec(() => {
|
|
describe("connect after a drop", () => {
|
|
it("throws while the previous parent link is open or still closing, and takes the next once it closed (race 4)", async () => {
|
|
const { client, links } = await linked((h) => void serveNotes(h));
|
|
const notes = await client.get(Notes, { session: "S1" }, context);
|
|
const next = (): void =>
|
|
client.connect(memoryLinkPair()[0], { principal: { kind: "server" }, codec: testCodec() }, context);
|
|
// The server's end closes now, the client's a microtask later.
|
|
links[0].close();
|
|
expect(next).toThrow("connect: this host has an open parent link");
|
|
await transported();
|
|
expect(notes.state.value).toEqual({ kind: "loading" });
|
|
expect(next).not.toThrow();
|
|
});
|
|
});
|
|
});
|
|
|
|
forEachCodec(() => {
|
|
describe("where a reference or an arriving hold goes (section 1's table)", () => {
|
|
/** A worker for session S1 serving `PlacedNotes`, a raw peer as its child, and one as its parent when `parent`. */
|
|
async function worker(options: { readonly parent?: boolean } = {}) {
|
|
const errors: HostError[] = [];
|
|
const host = createHost<TestPrincipal>({
|
|
address: Session({ id: "S1" }),
|
|
principal: { kind: "worker", session: "S1" },
|
|
onError: (error) => errors.push(error),
|
|
});
|
|
let parent: RawPeer | undefined;
|
|
if (options.parent === true) {
|
|
const [a, b] = memoryLinkPair();
|
|
host.connect(a, { principal: { kind: "server" }, codec: testCodec() }, context);
|
|
parent = new RawPeer(b);
|
|
}
|
|
await host.load(
|
|
[{ name: "@test/w", facet: defineFacet<TestPrincipal>((h) => serveNotes(h, PlacedNotes)) }],
|
|
context,
|
|
);
|
|
const [a, b] = memoryLinkPair();
|
|
host.accept(a, { principal: { kind: "owner" }, codec: testCodec() }, context);
|
|
return { host, child: new RawPeer(b), parent, errors };
|
|
}
|
|
|
|
/** Sends a hold of `PlacedNotes` (or `type`) for `session`, naming the session `address`. */
|
|
function holdOf(peer: RawPeer, session: string, address: string, type: string = PlacedNotes.name, hold = 0) {
|
|
peer.send({
|
|
kind: "hold",
|
|
hold,
|
|
type,
|
|
version: PlacedNotes.version,
|
|
fingerprint: PlacedNotes.fingerprint,
|
|
params: jsonPart({ session }),
|
|
address: { kind: "session", params: { id: address } },
|
|
});
|
|
}
|
|
|
|
const misplaced = {
|
|
code: "invalid",
|
|
reason: "host_address",
|
|
message: "app.placed-notes: placed on session, this host is session|S1",
|
|
};
|
|
|
|
it("binds a hold of a type served here that names this host's address", async () => {
|
|
const { child } = await worker();
|
|
holdOf(child, "S1", "S1");
|
|
expect((await child.next("bound")).holds[0]).toMatchObject({ hold: 0 });
|
|
});
|
|
|
|
it("keeps a hold naming this host whose params the code here places elsewhere waiting: invalid/host_address (code skew, M§1)", async () => {
|
|
const { child } = await worker();
|
|
holdOf(child, "S2", "S1");
|
|
expect(await child.next("failed")).toMatchObject({
|
|
hold: 0,
|
|
cause: { kind: "waiting", error: { kind: "chord", error: misplaced } },
|
|
});
|
|
// A refused hold is kept (a publication or a retry starts it again): the next hold takes the next id.
|
|
holdOf(child, "S2", "S1", PlacedNotes.name, 1);
|
|
expect(await child.next("failed")).toMatchObject({ hold: 1, cause: { kind: "waiting" } });
|
|
});
|
|
|
|
it("places a hold by the address it names, never by the name it serves (a relay never captures it)", async () => {
|
|
const { child, parent } = await worker({ parent: true });
|
|
holdOf(child, "S2", "S2");
|
|
const relayed = await parent!.next("hold");
|
|
expect(relayed).toMatchObject({ type: PlacedNotes.name, address: { kind: "session", params: { id: "S2" } } });
|
|
holdOf(child, "S1", "S2", PlacedNotes.name, 1);
|
|
expect(await parent!.next("hold")).toMatchObject({ hold: 1 });
|
|
});
|
|
|
|
it("sends a reference made here whose host address is another host over the parent link, though it serves the type", async () => {
|
|
const { host, parent } = await worker({ parent: true });
|
|
const elsewhere = host.hold(PlacedNotes, { session: "S2" }, context);
|
|
const hold = await parent!.next("hold");
|
|
expect(hold).toMatchObject({ type: PlacedNotes.name, address: { kind: "session", params: { id: "S2" } } });
|
|
expect(elsewhere.state.value).toEqual({ kind: "loading" });
|
|
expect((await host.get(PlacedNotes, { session: "S1" }, context)).variant).toBe("editor");
|
|
});
|
|
|
|
it("sends a hold it cannot serve or route over the parent link, unparsed", async () => {
|
|
const { child, parent } = await worker({ parent: true });
|
|
holdOf(child, "S9", "S9", "app.elsewhere");
|
|
const relayed = await parent!.next("hold");
|
|
expect(relayed).toMatchObject({ type: "app.elsewhere" });
|
|
expect(partValue(relayed.params)).toEqual({ session: "S9" });
|
|
});
|
|
|
|
it("never sends a hold from the parent back to it: unavailable (race 8)", async () => {
|
|
const { parent } = await worker({ parent: true });
|
|
holdOf(parent!, "S9", "S9", "app.elsewhere");
|
|
expect(await parent!.next("failed")).toMatchObject({
|
|
hold: 0,
|
|
cause: {
|
|
kind: "broken",
|
|
error: { kind: "chord", error: { code: "unavailable", message: "app.elsewhere is not served here" } },
|
|
},
|
|
});
|
|
expect(parent!.kinds()).toEqual(["hello"]);
|
|
});
|
|
|
|
it("refuses a hold it cannot serve or route without a parent link: unavailable", async () => {
|
|
const { child } = await worker();
|
|
holdOf(child, "S9", "S9", "app.elsewhere");
|
|
expect(await child.next("failed")).toMatchObject({
|
|
cause: { kind: "broken", error: { kind: "chord", error: { code: "unavailable" } } },
|
|
});
|
|
});
|
|
|
|
it("fails a reference broken with what its hostAddress throws", async () => {
|
|
const Throwing = remote("app.throwing", {
|
|
version: 1,
|
|
params: {},
|
|
hostAddress: () => {
|
|
throw new Error("hostAddress threw");
|
|
},
|
|
variants: { v: { state: { n: int() } } },
|
|
});
|
|
const client = createHost<TestPrincipal>(noErrors);
|
|
await client.load([], context);
|
|
expect(await rejection(client.get(Throwing, {}, context))).toMatchObject({
|
|
message: "app.throwing: hostAddress threw",
|
|
});
|
|
expect(client.hold(Throwing, {}, context).state.value).toMatchObject({ kind: "failed" });
|
|
});
|
|
});
|
|
});
|
|
|
|
describe("the host's store", () => {
|
|
/** A store on `:memory:`, as Chord's layers see it. */
|
|
async function memoryStore(): Promise<ChordStore> {
|
|
return (await openStore(sqliteBackend(await openNodeSqliteDatabase(":memory:")), context)) as ChordStore;
|
|
}
|
|
|
|
it("belongs to one open host at a time (rule 7)", async () => {
|
|
const store = await memoryStore();
|
|
const first = createHost({ store, onError: () => {} });
|
|
expect(() => createHost({ store, onError: () => {} })).toThrow("store: already attached to an open host");
|
|
await first.close(context);
|
|
await createHost({ store, onError: () => {} }).close(context);
|
|
await store.close(context);
|
|
});
|
|
|
|
it("reports storage_failed once and closes the host; the app still closes the store", async () => {
|
|
const { store, fail } = await faultyStore();
|
|
const errors: HostError[] = [];
|
|
const host = createHost({ store, onError: (error) => errors.push(error) });
|
|
await host.load([], context);
|
|
const write: StagedWrite = { order: "write", apply: async () => [] };
|
|
const fence = { admit: () => true, adopt: () => {} };
|
|
// A failed `COMMIT`: an unknown outcome (D4).
|
|
fail("COMMIT");
|
|
await expect(store.commit([write], fence, context)).rejects.toMatchObject({ reason: "storage_failed" });
|
|
await expect(store.commit([write], fence, context)).rejects.toMatchObject({ reason: "storage_failed" });
|
|
// The host closed itself; the app's own close returns that closing.
|
|
expect((host as ChordHost).closing).toBe(true);
|
|
await host.close(context);
|
|
expect(errors.map(({ kind }) => kind)).toEqual(["storage_failed"]);
|
|
await store.close(context);
|
|
});
|
|
});
|
|
|
|
/** A plugin's facet serving `Info` with the title `title` (a new facet value each call: a changed plugin). */
|
|
function infoFacet(title: string) {
|
|
return defineFacet((h) => h.serve(Info, () => ({ title: constant(title) })));
|
|
}
|
|
|
|
describe("reload: the API (slice 7b §1)", () => {
|
|
it("throws before load resolved and rejects released/closed after close", async () => {
|
|
const host = createHost(noErrors);
|
|
expect(() => host.reload([], context)).toThrow("reload runs after load");
|
|
await host.load([], context);
|
|
await host.close(context);
|
|
expect(await rejection(host.reload([], context))).toMatchObject({ code: "released", reason: "closed" });
|
|
});
|
|
|
|
it("reports plugins added, changed and removed; the same facet object is unchanged and untouched", async () => {
|
|
const core = infoFacet("core");
|
|
const host = createHost(noErrors);
|
|
const kept = defineFacet((h) => h.serve(Counter, () => ({ count: constant(1), bump: async () => null })));
|
|
await host.load(
|
|
[
|
|
{ name: "@app/core", facet: kept },
|
|
{ name: "@app/info", facet: core },
|
|
],
|
|
context,
|
|
);
|
|
const counter = await host.get(Counter, {}, context);
|
|
const report = await host.reload(
|
|
[
|
|
{ name: "@app/core", facet: kept },
|
|
{ name: "@app/page", facet: defineFacet((h) => h.serve(Page, ({ id }) => ({ id: constant(id) }))) },
|
|
],
|
|
context,
|
|
);
|
|
expect(report.plugins).toEqual({ added: ["@app/page"], changed: [], removed: ["@app/info"] });
|
|
expect(report.objects).toEqual([{ object: "app.info", kind: "ended" }]);
|
|
// Untouched: the same reference stays live on the same object.
|
|
expect(counter.state.value).toEqual({ kind: "live" });
|
|
expect((await host.get(Page, { id: 3 }, context)).id.value).toBe(3);
|
|
await host.close(context);
|
|
});
|
|
|
|
it("runs reloads one after another", async () => {
|
|
const host = createHost(noErrors);
|
|
await host.load([{ name: "@app/info", facet: infoFacet("a") }], context);
|
|
const order: string[] = [];
|
|
const first = host.reload([{ name: "@app/info", facet: infoFacet("b") }], context).then(() => order.push("b"));
|
|
const second = host.reload([{ name: "@app/info", facet: infoFacet("c") }], context).then(() => order.push("c"));
|
|
await Promise.all([first, second]);
|
|
expect(order).toEqual(["b", "c"]);
|
|
expect((await host.get(Info, {}, context)).title.value).toBe("c");
|
|
await host.close(context);
|
|
});
|
|
|
|
it("an abort before the commit rejects aborted and changes nothing", async () => {
|
|
const host = createHost(noErrors);
|
|
await host.load([{ name: "@app/info", facet: infoFacet("a") }], context);
|
|
const { context: aborted, cancel } = withCancel(context);
|
|
cancel(new Error("stop"));
|
|
expect(await rejection(host.reload([{ name: "@app/info", facet: infoFacet("b") }], aborted))).toMatchObject({
|
|
code: "aborted",
|
|
});
|
|
expect((await host.get(Info, {}, context)).title.value).toBe("a");
|
|
await host.close(context);
|
|
});
|
|
|
|
it("an abort after the commit rejects aborted naming what the commit applied, which stands", async () => {
|
|
const host = createHost(noErrors);
|
|
await host.load(
|
|
[
|
|
{ name: "@app/info", facet: infoFacet("a") },
|
|
{ name: "@app/gone", facet: defineFacet(() => {}) },
|
|
],
|
|
context,
|
|
);
|
|
const info = await host.get(Info, {}, context);
|
|
const { context: aborting, cancel } = withCancel(context);
|
|
// The abort lands right after the commit (its publication batch), while the reload waits for its report.
|
|
const { batch } = host as ChordHost;
|
|
const commit = batch.commit.bind(batch);
|
|
const hook = vi.spyOn(batch, "commit").mockImplementation((adopt, ctx) => {
|
|
commit(adopt, ctx);
|
|
if (ctx === aborting) cancel(new Error("stop"));
|
|
});
|
|
const reloading = host.reload(
|
|
[
|
|
{ name: "@app/info", facet: infoFacet("b") },
|
|
{ name: "@app/new", facet: defineFacet(() => {}) },
|
|
],
|
|
aborting,
|
|
);
|
|
const error = await rejection(reloading);
|
|
hook.mockRestore();
|
|
expect(error).toMatchObject({
|
|
code: "aborted",
|
|
details: { committed: { added: ["@app/new"], changed: ["@app/info"], removed: ["@app/gone"], failed: [] } },
|
|
});
|
|
await flushed();
|
|
expect(info.title.value).toBe("b");
|
|
await host.close(context);
|
|
});
|
|
|
|
it("reports a setup's or a factory's throw as failed, naming the plugin, which stays on its old code", async () => {
|
|
const host = createHost(noErrors);
|
|
await host.load([{ name: "@app/info", facet: infoFacet("a") }], context);
|
|
const info = await host.get(Info, {}, context);
|
|
const throwing = defineFacet(() => {
|
|
throw new Error("setup threw");
|
|
});
|
|
const failedOf = async (facets: { readonly name: string; readonly facet: typeof throwing }[]) => {
|
|
const report = await host.reload(facets, context);
|
|
return report.failed.map(({ plugin, error }) => [plugin, (error as Error).message]);
|
|
};
|
|
expect(await failedOf([{ name: "@app/info", facet: throwing }])).toEqual([
|
|
["@app/info", 'plugin "@app/info": setup threw'],
|
|
]);
|
|
const factoryThrows = defineFacet((h) =>
|
|
h.serve(Info, () => {
|
|
throw new Error("factory threw");
|
|
}),
|
|
);
|
|
expect(await failedOf([{ name: "@app/info", facet: factoryThrows }])).toEqual([
|
|
["@app/info", 'plugin "@app/info", app.info: factory threw'],
|
|
]);
|
|
// A type another plugin serves already fails the plugin with new code; an added one is left out.
|
|
const twice = [
|
|
{ name: "@app/info", facet: infoFacet("b") },
|
|
{ name: "@app/other", facet: infoFacet("c") },
|
|
];
|
|
const report = await host.reload(twice, context);
|
|
expect(report.plugins).toEqual({ added: [], changed: ["@app/info"], removed: [] });
|
|
expect(report.failed.map(({ plugin, error }) => [plugin, (error as Error).message])).toEqual([
|
|
["@app/other", 'plugin "@app/other": app.info is served twice (plugin "@app/info")'],
|
|
]);
|
|
// Restarted: the new attempt goes live once the old one ended.
|
|
await flushed();
|
|
expect(info.title.value).toBe("b");
|
|
expect(info.state.value).toEqual({ kind: "live" });
|
|
await host.close(context);
|
|
});
|
|
});
|
|
|
|
describe("reload: the commit (slice 7b §3)", () => {
|
|
it("serves added types, drops removed ones and their presence lists, and wakes requests per name", async () => {
|
|
const host = createHost(noErrors) as unknown as ChordHost;
|
|
await host.load([{ name: "@app/info", facet: infoFacet("a") }], context);
|
|
const page = host.hold(Page, { id: 1 }, context);
|
|
const other = host.hold(Core, {}, context);
|
|
await flushed();
|
|
expect(page.state.value).toMatchObject({ kind: "failed", cause: { kind: "waiting" } });
|
|
const presence = host.references(Info);
|
|
const otherStates = statesOf(await other.state.watch(context));
|
|
await host.reload(
|
|
[
|
|
{ name: "@app/info", facet: infoFacet("a") },
|
|
{ name: "@app/page", facet: defineFacet((h) => h.serve(Page, ({ id }) => ({ id: constant(id) }))) },
|
|
],
|
|
context,
|
|
);
|
|
await flushed();
|
|
expect(page.state.value).toEqual({ kind: "live" });
|
|
expect(page.id.value).toBe(1);
|
|
// Not a name the reload serves: still waiting, not checked again.
|
|
expect(other.state.value).toMatchObject({ kind: "failed", cause: { kind: "waiting" } });
|
|
expect(otherStates).toEqual([]);
|
|
await host.reload(
|
|
[{ name: "@app/page", facet: defineFacet((h) => h.serve(Page, ({ id }) => ({ id: constant(id) }))) }],
|
|
context,
|
|
);
|
|
expect(host.references(Info)).not.toBe(presence);
|
|
await host.close(context);
|
|
});
|
|
|
|
it("replaces a reloaded registry's provider and its plugin's contributions; a removed provider's readers wait", async () => {
|
|
const Names = defineRegistry<string, string>("app.names", { version: 1 });
|
|
const provider = (separator: string) =>
|
|
defineFacet((h, self) => {
|
|
h.serve(Names, { compose: (table) => table.map(({ value }) => value).join(separator) });
|
|
self.contribute(Names, "own");
|
|
});
|
|
const contributor = (name: string) => defineFacet((_h, self) => self.contribute(Names, name));
|
|
const host = createHost(noErrors);
|
|
await host.load(
|
|
[
|
|
{ name: "@app/names", facet: provider(",") },
|
|
{ name: "@app/x", facet: contributor("x") },
|
|
],
|
|
context,
|
|
);
|
|
const names = await host.registry(Names, context);
|
|
expect(names.value).toBe("own,x");
|
|
await host.reload(
|
|
[
|
|
{ name: "@app/names", facet: provider("+") },
|
|
{ name: "@app/x", facet: contributor("y") },
|
|
],
|
|
context,
|
|
);
|
|
expect(names.value).toBe("own+y");
|
|
await host.reload([{ name: "@app/x", facet: contributor("z") }], context);
|
|
expect(() => names.value).toThrow("no plugin serves it here");
|
|
await host.reload(
|
|
[
|
|
{ name: "@app/names", facet: provider(";") },
|
|
{ name: "@app/x", facet: contributor("z") },
|
|
],
|
|
context,
|
|
);
|
|
expect(names.value).toBe("own;z");
|
|
await host.close(context);
|
|
});
|
|
});
|
|
|
|
/** A worker's title per session (v1: a string). */
|
|
const Title = remote("app.title", {
|
|
version: 1,
|
|
params: { session: str() },
|
|
hostAddress: ({ session }) => Session({ id: session }),
|
|
variants: { v: { state: { title: str() } } },
|
|
});
|
|
|
|
/** `Title` v2: an object; v1 consumers map back through `accepts`. */
|
|
const TitleV2 = evolve(Title, (e) => e.state("v", "title", obj({ text: str() })));
|
|
|
|
describe("reload: registries' older readers (slice 7b §3)", () => {
|
|
it("a reader of an older version follows the new provider's accepts[old].value; one it no longer serves waits", async () => {
|
|
const NamesV1 = defineRegistry<string, string>("app.names", { version: 1 });
|
|
const NamesV2 = defineRegistry<string, string[]>("app.names", { version: 2 });
|
|
const provider = (map: ((names: string[]) => string) | undefined) =>
|
|
defineFacet((h, self) => {
|
|
h.serve(NamesV2, {
|
|
compose: (table) => table.map(({ value }) => value),
|
|
...(map === undefined ? {} : { accepts: { 1: { value: map } } }),
|
|
});
|
|
self.contribute(NamesV2, "a");
|
|
});
|
|
const host = createHost(noErrors);
|
|
await host.load([{ name: "@app/names", facet: provider((names) => names.join(",")) }], context);
|
|
const old = await host.registry(NamesV1, context);
|
|
expect(old.value).toBe("a");
|
|
await host.reload([{ name: "@app/names", facet: provider((names) => `${names.join("+")}!`) }], context);
|
|
expect(old.value).toBe("a!");
|
|
await host.reload([{ name: "@app/names", facet: provider(undefined) }], context);
|
|
expect(() => old.value).toThrow("version 1 is not served here");
|
|
await host.close(context);
|
|
});
|
|
});
|
|
|
|
describe("reload across hosts, in the app's order (slice 7b §11)", () => {
|
|
it("server, then worker, then tab: an old tab is mapped, then waits for its version, then reloads its own code", async () => {
|
|
const serveV1 = (h: FacetHost<TestPrincipal>) =>
|
|
h.serve(Title, () => ({ variant: "v", members: { title: constant("one") } }));
|
|
const serveV2 = (text: string, accept: boolean) => (h: FacetHost<TestPrincipal>) =>
|
|
h.serve(
|
|
TitleV2,
|
|
() => ({ variant: "v", members: { title: constant({ text }) } }),
|
|
accept ? { accepts: { 1: { state: { v: { title: (current) => current.text } } } } } : {},
|
|
);
|
|
const workerSet = (setup: (h: FacetHost<TestPrincipal>) => void) => [
|
|
{
|
|
name: "@test/worker",
|
|
facet: defineFacet<TestPrincipal>((h) => {
|
|
serveNotes(h, PlacedNotes);
|
|
setup(h);
|
|
}),
|
|
},
|
|
];
|
|
const { tab, server, workers } = await relayed({ worker: serveV1 });
|
|
const old = tab.hold(Title, { session: "S1" }, context);
|
|
await transported();
|
|
expect(old.title.value).toBe("one");
|
|
const worker = workers.get("session|S1")!;
|
|
// The server installs first: nothing of Title changes there.
|
|
await server.reload([{ name: "@test/server", facet: defineFacet<TestPrincipal>(() => {}) }], context);
|
|
// The worker installs v2 accepting v1: the old tab is mapped.
|
|
await worker.reload(workerSet(serveV2("two", true)), context);
|
|
await transported();
|
|
expect(old.state.value).toEqual({ kind: "live" });
|
|
expect(old.title.value).toBe("two");
|
|
// A worker install without the adapter: the old tab waits for its version (a remote failure is its reads' error).
|
|
await worker.reload(workerSet(serveV2("three", false)), context);
|
|
await transported();
|
|
expect(old.state.value).toMatchObject({
|
|
kind: "failed",
|
|
cause: { kind: "waiting", error: { code: "contract_mismatch" } },
|
|
});
|
|
// The tab reloads its own code: its new hold (v2) binds; the old one is released.
|
|
const fresh = tab.hold(TitleV2, { session: "S1" }, context);
|
|
old.release();
|
|
await transported();
|
|
expect(fresh.title.value).toEqual({ text: "three" });
|
|
await server.close(context);
|
|
});
|
|
});
|
|
|
|
describe("reload: an abort while staging (fix round 1)", () => {
|
|
it("does not commit when the context is aborted by the synchronous setup", async () => {
|
|
const T = local("astra.cancel", { version: 1, params: {}, members: { n: int() } });
|
|
const host = createHost({ onError: () => {} });
|
|
const original = defineFacet((h) => h.serve(T, () => ({ n: constant(1) })));
|
|
await host.load([{ name: "@astra/cancel", facet: original }], context);
|
|
const ref = await host.get(T, {}, context);
|
|
const abort = new AbortController();
|
|
const changed = defineFacet((h) => {
|
|
abort.abort();
|
|
h.serve(T, () => ({ n: constant(2) }));
|
|
});
|
|
await host
|
|
.reload([{ name: "@astra/cancel", facet: changed }], withAbortSignal(abort.signal, context))
|
|
.catch(() => {});
|
|
await flushed();
|
|
const actual = ref.n.value;
|
|
ref.release();
|
|
await host.close(context);
|
|
expect(actual).toBe(1);
|
|
});
|
|
});
|
|
|
|
describe("who a span was for (principalAttributes)", () => {
|
|
const Echo = local("app.echo", { version: 1, params: {}, members: { echo: method(str(), str()) } });
|
|
|
|
/** A host acting as `{ kind: "user", id: "u1", name: "Ada" }` serving `Echo`, recording to `telemetry`. */
|
|
async function host(
|
|
telemetry: InMemoryTelemetryContext,
|
|
principalAttributes?: (p: { kind: string }) => Readonly<SpanAttributes>,
|
|
) {
|
|
const made = createHost({
|
|
principal: { kind: "user", id: "u1", name: "Ada" },
|
|
telemetry,
|
|
...(principalAttributes === undefined ? {} : { principalAttributes }),
|
|
onError: () => {},
|
|
});
|
|
await made.load(
|
|
[{ name: "@test/echo", facet: defineFacet((h) => h.serve(Echo, () => ({ echo: async (text) => text }))) }],
|
|
context,
|
|
);
|
|
const echo = await made.get(Echo, {}, context);
|
|
await telemetry.startSpan({ name: "work" }, (span) => echo.echo("hi", withTelemetry(span, context)));
|
|
return telemetry.getSpans().find(({ name }) => name === "chord.method")!.attributes;
|
|
}
|
|
|
|
it("records a method's principal key as `chord.principal` by default", async () => {
|
|
expect(await host(new InMemoryTelemetryContext())).toMatchObject({
|
|
"chord.principal": '{"id":"u1","kind":"user","name":"Ada"}',
|
|
});
|
|
});
|
|
|
|
it("records the hook's attributes in its place, called once per recorded span", async () => {
|
|
const hook = vi.fn((principal: { kind: string }) => ({ "chord.principal": principal.kind, "user.name": "Ada" }));
|
|
const attributes = await host(new InMemoryTelemetryContext(), hook);
|
|
expect(attributes).toMatchObject({ "chord.principal": "user", "user.name": "Ada" });
|
|
expect(hook).toHaveBeenCalledTimes(1);
|
|
});
|
|
|
|
it("records nothing of the principal when the hook returns {}", async () => {
|
|
const attributes = await host(new InMemoryTelemetryContext(), () => ({}));
|
|
expect(Object.keys(attributes).filter((key) => key === "chord.principal" || key === "user.name")).toEqual([]);
|
|
expect(attributes).toMatchObject({ "principal.kind": "user" });
|
|
});
|
|
|
|
it("never calls the hook when nothing records (the NOOP adapter)", async () => {
|
|
const hook = vi.fn(() => ({ "chord.principal": "x" }));
|
|
const silent = createHost({ principal: { kind: "user" }, principalAttributes: hook, onError: () => {} });
|
|
await silent.load(
|
|
[{ name: "@test/echo", facet: defineFacet((h) => h.serve(Echo, () => ({ echo: async (text) => text }))) }],
|
|
context,
|
|
);
|
|
await (await silent.get(Echo, {}, context)).echo("hi", context);
|
|
expect(hook).not.toHaveBeenCalled();
|
|
});
|
|
|
|
it("records a call span's principal: whom its reference acts for", async () => {
|
|
const Greeter = remote("app.greeter", {
|
|
version: 1,
|
|
params: {},
|
|
hostAddress: () => hostKind("server")(),
|
|
variants: { main: { methods: { greet: method(str(), str()) } } },
|
|
});
|
|
const server = createHost<{ kind: string; id?: string }>({
|
|
address: hostKind("server")(),
|
|
principal: { kind: "server" },
|
|
onError: () => {},
|
|
});
|
|
await server.load(
|
|
[
|
|
{
|
|
name: "@test/greeter",
|
|
facet: defineFacet((h) =>
|
|
h.serve(Greeter, () => ({ variant: "main", members: { greet: async (name) => `hi ${name}` } })),
|
|
),
|
|
},
|
|
],
|
|
context,
|
|
);
|
|
const telemetry = new InMemoryTelemetryContext();
|
|
const client = createHost<{ kind: string; id?: string }>({
|
|
principal: { kind: "user", id: "u1" },
|
|
telemetry,
|
|
principalAttributes: (principal) => ({ "chord.principal": `${principal.kind}:${principal.id}` }),
|
|
onError: () => {},
|
|
});
|
|
await client.load([], context);
|
|
const [a, b] = memoryLinkPair();
|
|
server.accept(a, { principal: { kind: "user", id: "u1" } }, context);
|
|
client.connect(b, { principal: { kind: "server" } }, context);
|
|
const greeter = await client.get(Greeter, {}, context);
|
|
await telemetry.startSpan({ name: "click" }, (span) => greeter.greet("Ada", withTelemetry(span, context)));
|
|
const call = telemetry.getSpans().find(({ name }) => name === "chord.call")!;
|
|
expect(call.attributes).toMatchObject({ "chord.principal": "user:u1" });
|
|
});
|
|
});
|
|
|
|
describe("the host's spans (slice 9)", () => {
|
|
it("traces load under its caller's span: plugins given and left out; host-lifetime activations link to it", async () => {
|
|
const telemetry = new InMemoryTelemetryContext();
|
|
const host = createHost({ telemetry, onError: () => {} });
|
|
const ok = defineFacet((h) => h.serve(Info, () => ({ title: constant("t") })));
|
|
const failing = defineFacet(() => {
|
|
throw new Error("setup");
|
|
});
|
|
await telemetry.startSpan({ name: "start" }, (span) =>
|
|
host.load(
|
|
[
|
|
{ name: "@test/ok", facet: ok },
|
|
{ name: "@test/failing", facet: failing },
|
|
],
|
|
withTelemetry(span, context),
|
|
),
|
|
);
|
|
const spans = telemetry.getSpans();
|
|
const load = spans.find(({ name }) => name === "chord.load")!;
|
|
expect(load.parentId).toBe(spans.find(({ name }) => name === "start")!.id);
|
|
expect(load.attributes).toEqual({ "chord.host": "client", plugins: 2, failed: 1 });
|
|
const activation = spans.find(({ name }) => name === "chord.activate")!;
|
|
expect(activation.links.map(({ context: carrier }) => carrier)).toEqual([load.context]);
|
|
});
|
|
|
|
it("traces a reload with one event per object's fate", async () => {
|
|
const telemetry = new InMemoryTelemetryContext();
|
|
const host = createHost({ telemetry, onError: () => {} });
|
|
const facet = (title: string) => defineFacet((h) => h.serve(Info, () => ({ title: constant(title) })));
|
|
await host.load([{ name: "@test/app", facet: facet("a") }], context);
|
|
await telemetry.startSpan({ name: "apply" }, (span) =>
|
|
host.reload([{ name: "@test/app", facet: facet("b") }], withTelemetry(span, context)),
|
|
);
|
|
const reload = telemetry.getSpans().find(({ name }) => name === "chord.reload")!;
|
|
expect(reload.attributes).toEqual({
|
|
"chord.host": "client",
|
|
plugins: 1,
|
|
added: 0,
|
|
changed: 1,
|
|
removed: 0,
|
|
failed: 0,
|
|
fates: 1,
|
|
});
|
|
expect(reload.events).toEqual([
|
|
{ name: "chord.reload.fate", attributes: { object: "app.info", fate: "restarted" } },
|
|
]);
|
|
});
|
|
|
|
it("counts the cost counters of its links' and roots' work on its adapter, under contexts without a span (review L6)", async () => {
|
|
const telemetry = new InMemoryTelemetryContext();
|
|
const { server, client } = await linked((h) => void serveNotes(h), { telemetry });
|
|
const notes = await client.get(Notes, { session: "S1" }, context);
|
|
if (notes.variant !== "editor") throw new Error("expected the editor");
|
|
await notes.add({ text: "a" }, context);
|
|
await server.root(async (ctx) => (await ctx.get(Notes, { session: "S1" })).release(), context);
|
|
const counted = new Set(telemetry.getMetrics().map(({ name }) => name));
|
|
expect(counted).toContain("chord.link.checks");
|
|
expect(counted).toContain("chord.data.validations");
|
|
expect(counted).toContain("chord.link.encodes");
|
|
});
|
|
|
|
it("records a revocation's count after it under the caller's span", async () => {
|
|
const telemetry = new InMemoryTelemetryContext();
|
|
const host = createHost<TestPrincipal, string>({
|
|
address: Server(),
|
|
principal: { kind: "server" },
|
|
telemetry,
|
|
revokedBy: (principal) => (principal.kind === "guest" ? [principal.session] : []),
|
|
onError: () => {},
|
|
});
|
|
await host.load([], context);
|
|
const [a] = memoryLinkPair();
|
|
host.accept(a, { principal: { kind: "guest", session: "S1" } }, context);
|
|
await telemetry.startSpan({ name: "admin" }, (span) => host.revoke("S1", withTelemetry(span, context)));
|
|
const revoke = telemetry.getSpans().find(({ name }) => name === "chord.revoke")!;
|
|
expect(revoke.attributes).toEqual({ "chord.host": "server", count: 1 });
|
|
host.revoke("S1", context);
|
|
expect(telemetry.getSpans().filter(({ name }) => name === "chord.revoke")).toHaveLength(1);
|
|
});
|
|
});
|