sscoble/pi-mono
1
0
Fork 0
mirror of https://github.com/badlogic/pi-mono.git synced 2026-11-05 23:37:25 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 4

feat(chord): calls in flight survive a loading unbind

Browse source 
A hold unbinding with loading (a restart, a parked swap, a live failure retrying) keeps its calls in flight; each
gets one answer from what it waits on (the instance it runs on, or the gate). Calls sent by the old method names
before the consumer learns of the loading wait at the gate too. failed and removal still reject.
This commit is contained in:
Mario Zechner 2026-11-04 15:12:13 +00:00
parent 395f3e0318
commit 6515a79772
8 changed files with 528 additions and 59 deletions
Download patch file Download diff file Expand all files Collapse all files

3
packages/chord/CHANGELOG.md
View file

@ -33,6 +33,7 @@
- Chord v2's `host.load` and `host.reload` throw for a plugin name that is no npm package name, before they change anything (it was listed in the report's `failed`).
- Chord v2 plugins import each other by relative path within a root (copied into the importer's script) and across repositories by package name through the package's `exports`, pinned in `dependencies`; slice 10's imports by plugin name (`@app/notes/contract`) are removed, and a relative path leaving the root is a build error.
- Chord v2's `host.reload` starts the objects' fates dependents first: removals, then the other fates in the reverse of the order the objects went live (the report keeps its order). A pure dependent reloaded with its provider (a view over a store in one plugin) binds the provider while it is still live and swaps instead of parking until the provider's new instance is live: its contributions are replaced in one batch and its holds over links never show `loading`. A dependent still parks when its new code needs the provider's new version, when its provider went live after it, or when it needs something not live.
- Chord v2 keeps calls over a link across a `loading` unbind (decision "Calls in flight survive a `loading` unbind"): a hold whose object restarts, whose swap parks or whose live failure retries no longer answers its calls in flight `unavailable`; each is answered by the instance it already runs on or by the new instance once it is live (or by its failure, `readyTimeout`, its end), and calls a consumer sent by the old names before it saw the `loading` wait the same way instead of `not_ready`. A `failed` unbind (a removed type, a dropped version, a swap parked on a failed dependency) still rejects them.
### Fixed
@ -40,7 +41,7 @@
- Chord v2's `processLock` on macOS spun until its context ended when a regular file sat at the lock's path (`ENOTSOCK`); it now fails closed naming the file, as for a socket file nobody listens on.
- Chord v2's heartbeat closed every link after the process had slept (a laptop's sleep) or blocked its event loop, before reading what the peer had sent meanwhile. Time the process did not run no longer counts as silence: after such a pause each open link pings at once and closes only if its peer stays silent for `close - ping` (15 s by default).
- Chord v2's workers no longer leave an address failed when a running worker loses its link (the heartbeat, a protocol error, the worker closing its own link): the worker is stopped, `worker_link_lost { address, error }` is reported, and its holders get a new worker. A second lost link within a minute of the last recovery still fails the address (public error `worker lost its link`). Crashes, `fatal`s and failed starts still stay failed until `retry` or a rollout.
- Chord v2's objects no longer answer a call `not_ready` during a reload while the caller sees them `live`: a call over a link to an object whose swap had parked (its new code waiting for a dependency) or whose restart had drained, sent while its hold was still bound to the old instance, got `not_ready: ... not live yet`. Such a call now waits for the object's new attempt (live, failed or ended), bounded by `readyTimeout` (`unavailable/park_timeout`); a parked attempt showing `failed { waiting }` rejects it with that cause, and a hold that leaves the old instance answers it `unavailable` with `loading`.
- Chord v2's objects no longer answer a call `not_ready` during a reload while the caller sees them `live`: a call over a link to an object whose swap had parked (its new code waiting for a dependency) or whose restart had drained, sent while its hold was still bound to the old instance, got `not_ready: ... not live yet`. Such a call now waits for the object's new attempt (live, failed or ended), bounded by `readyTimeout` (`unavailable/park_timeout`); a parked attempt showing `failed { waiting }` rejects it with that cause; it stays when the hold then leaves the old instance with `loading`.
## [1.5.0] - 2026-11-02

26
packages/chord/docs/design.md
View file

@ -391,7 +391,7 @@ host.route(Session, {
nothing is refused for being "too many" (K16). A slot is freed once per call: by its `result` or `error`, its
`abort` (when sent, at the caller; when received, at the provider), its hold's `release`, `gone`, its origin's
`closed` or the link closing. A call still queued when its hold rebinds (`loading`) is rejected `unavailable` (it may
name a method of the old variant). A provider seeing more than N running calls of one origin closes the link
name a method of the old variant); one already sent is answered by the provider (a `loading` unbind keeps it, below). A provider seeing more than N running calls of one origin closes the link
(protocol error); a call naming a hold it does not know is ignored before it is counted (Q1). **There is no limit on
calls in progress per link** (owner, 2026-11-03; Chord v1's `maxCalls` had no counterpart): the call window N per origin
and the part limits (section 22) are the bounds; 200 concurrent calls over one socket all run, N at a time in
@ -1974,8 +1974,9 @@ type ReloadReport = {
object activates (a swap, parked or not, a restart after its drain), until the object is live, failed or ended;
a parked attempt showing `failed { waiting }` rejects it with that cause. A call never gets `not_ready` there: a
hold over a link stays bound while the old instance still changes, so its consumer sees `live` and sends. When the
hold then leaves the old instance (a parked swap's retirement), it tells the consumer `loading` and answers the
calls still in flight `unavailable` (§22), as for any object leaving `live`. The new attempt's requests are made and bound
hold then leaves the old instance (a parked swap's retirement), it tells the consumer `loading` and keeps the calls
still in flight: they go on waiting at the gate (§22, decision "Calls in flight survive a `loading` unbind"). The
new attempt's requests are made and bound
before the old attempt's are released (a request equal to an old one binds to the same target with no gap, M§3.9).
Running calls finish on the old attempt with its handles valid, except R15: an old call touching a document whose
version rose throws once (`released/version`), the document intact, the call not resent; the same for a table whose
@ -2253,9 +2254,22 @@ validation, no clone of snapshots or ops.
detaches the hold's sources, drops its unsent pending ops and a queued `bound`, and keeps the ack slots of `bound`
and `state` messages already sent until their `ack` or the link's loss. A reload sends it when the object's type is
removed (`not_served`), its version is no longer accepted (`version`), or a swap parks on a failed dependency; a
restart or a parked swap still loading sends `loading`; both are followed by a fresh `bound` when the object is live
again (applied as a re-binding). A swap in place keeps the hold bound and sends it a fresh `bound` once the new
instance is live (audit A9, 2026-10-30: one rebind path; a tab's watchers may get a frame for an unchanged member).
restart, a parked swap still loading or a live failure already retrying sends `loading`; both are followed by a fresh
`bound` when the object is live again (applied as a re-binding). A swap in place keeps the hold bound and sends it a
fresh `bound` once the new instance is live (audit A9, 2026-10-30: one rebind path; a tab's watchers may get a frame
for an unchanged member).
- **A `loading` unbind keeps the calls in flight** (decision "Calls in flight survive a `loading` unbind"): the
provider unbinds the hold as above but rejects none of its calls, and the consumer keeps them (only its calls still
waiting for a slot reject `unavailable`). Each kept call gets exactly one answer from what it already waits on: the
instance it runs on (its result within `closeTimeout`, else `released` at the seal; a hold binding afresh after a swap
in place may unbind while its live instance still runs calls) or the gate (the new instance's result, the parked
cause, `unavailable` after `readyTimeout`, the object's end); an `abort`, a `release`, `gone` or the link's close
ends it as any call. Until its fresh `bound` is sent the hold takes calls by the names it was bound with (sent before
its consumer learnt of the `loading`): they wait at the gate too, never `not_ready`; after a `failed` it takes none.
The kept call runs by name: a method the new variant lacks fails `invalid/member`; a version the new code no longer
binds unbinds the hold (`failed`, `contract_mismatch`), which rejects its calls before the gate lets them run. Kept
calls hold their call-window slots meanwhile. A relay forwards `loading` and keeps its forwarded calls (its own
`loading` after an upstream drop rejects them first).
---

12
packages/chord/docs/slices.md
View file

@ -200,18 +200,6 @@ deleted (the slice's spec and code are the record). Moved here from `carry-over.
- **`rebuild` of tables** (key and column-type changes): `evolveTable` has no edit for them; build one only when a
real case needs it. (Slice 6a's second spec review; out of slice 7b's scope, `slice-7b.md` §12.)
- **Calls surviving a `loading` unbind** (decision "Swaps: dependents first; calls during a swap never see
`not_ready`", option B): today a call over a link already waiting at the gate when its object's swap parks is
answered `unavailable` when the hold leaves the old instance (`Hold.#unbind` → `Provider.unbound` rejects every
call in flight on the hold, with `loading`; consistent, rare after the dependents-first order). Sketch, about 5
lines: on an unbind for `loading` (no error: the object activates), `Provider.unbound` unbinds the hold but rejects
none of its calls: by then the old attempt is always sealed (`failLive`, `#restart` and `#retire` seal before
`#stopFollowing`), so its running calls were rejected `released` by the seal fence, and the calls left in flight
wait at the gate; they are answered after the fresh `bound` (or by the gate's `readyTimeout`). Checks it needs: the
consumer keeps in-flight calls across `loading` (`Consumer.#loading` rejects only calls still waiting for a slot,
which may name a method of the old variant; in-flight ones wait for the provider's reply), a relay forwarding
`loading` keeps its mapped call ids, a reply on a hold that is pending or binding is accepted, and a variant change
across the rebind (the call was decoded by the old names). (Owner, 2026-11-04: separate change, its own risk.)
### Later: live queries (when a case needs it)

8
packages/chord/src/remote/README.md
View file

@ -95,7 +95,10 @@ within each link's limits, compression at the origin, activation chains and revo
failed already; the next attempt (or the parked object's next `loading`) is announced with `loading`; the hold stays,
so a `retry` (from any holder) renews even a broken object for every holder (a parked object passes it on, one walk
of this host with one hop less of the message's budget). After a swap in place it gets a fresh `bound` once the new
instance is live, answering calls by the names it was bound with until that is sent (the window may hold it). Live,
instance is live, answering calls by the names it was bound with until that is sent (the window may hold it). An
object leaving `live` unbinds it with `failed` (its calls in flight rejected) or `loading`: then its calls in flight
stay, each answered by the instance it runs on or by the object's gate, and calls by the names it was bound with
still arrive and wait at the gate until its fresh `bound` (decision "Calls in flight survive a `loading` unbind"). Live,
it binds through the window (`boundEntry`: the variant and each member's value as a part, names as indexes into the
bound version). Its members' frames arrive as `Subscriber` calls inside the batch; at its
end the hold's entry lists the members that changed with their ops, each encoded with that member's own Delta
@ -138,7 +141,8 @@ within each link's limits, compression at the origin, activation chains and revo
calls wait in one FIFO queue, sent in order as replies free slots; it does not preempt running calls. A relay
forwards its downstream calls at once, so only this host's own calls wait here. A slot is freed once per call: its
`result` or `error`, its `abort` (when sent), its hold's `release` or `gone`, or the link's close. A call still
waiting when its hold gets `loading` is rejected `unavailable` (it may name a method of the old variant).
waiting when its hold gets `loading` is rejected `unavailable` (it may name a method of the old variant); one sent
already waits for its reply.
- `Path` (the parent, or one route address): its current link and its holders (`RemoteObject`s and `RelayedHold`s),
held on that link and again on every next one; when the link drops, every holder is told (`dropped`). `changed`
tells its owner (a route) what happened without a link: a holder `attached`, the link `dropped`, the last holder

3
packages/chord/src/remote/consumer.ts
View file

@ -299,7 +299,8 @@ export class ConsumerSide {
/**
* `loading`: the hold's object, or its relay's upstream, starts over; its calls still waiting for a slot reject
* `unavailable` (they may name a method of the old variant).
* `unavailable` (they may name a method of the old variant), its calls in flight wait for their reply (a provider's
* `loading` keeps them; a relay's upstream drop answered them before).
*/
#loading(hold: number): void {
const holder = this.#entries.get(hold);

67
packages/chord/src/remote/hold.ts
View file

@ -4,14 +4,14 @@
// taken in one synchronous step when a slot of the window is free, and the ops of its members per batch (sent at
// once, or pending while the window is full), each member with its own Delta encoder. Across a reload (slice 7b §6) a
// bound hold is kept: a swap in place sends it a fresh `bound` once the new instance is live (it answers calls
// meanwhile), an object leaving `live` unbinds it (`loading` or `failed`, then a fresh `bound`), a version no longer
// bound or a removed type leaves it waiting unbound. A hold of an older version follows its object's adapted members
// (Q10).
// meanwhile), an object leaving `live` unbinds it (`loading` or `failed`, then a fresh `bound`; a `loading` one keeps
// its calls in flight and takes more by its old names), a version no longer bound or a removed type leaves it waiting
// unbound. A hold of an older version follows its object's adapted members (Q10).
import type { TelemetryContext } from "@earendil-works/pi-telemetry";
import type { Context } from "../context/index.ts";
import { type VariantData, type VariantNames, variantNames, variantOrder, variantsAt } from "../contracts/internal.ts";
import { ChordError, type FailureCause, isChordError, type Principal } from "../data/index.ts";
import { type ChordError, type FailureCause, isChordError, type Principal } from "../data/index.ts";
import { asChordError } from "../data/internal.ts";
import { type Encoder, encoder, type Op, PendingOps } from "../delta/index.ts";
import {
@ -68,14 +68,15 @@ type Followed = {
/**
* Where a hold stands: `unplaced` in the unbound index until the host's publication places it; `pending` until its
* object is live (the activation, parking and retries; `toldFailed` once the consumer was told of a failure, so the
* next attempt is announced with `loading`); `binding` while its object is live and its `bound` waits for a slot
* (after a swap in place with the names it was bound with, `previous`: the consumer still holds it bound, so its calls
* are answered by those names, design §22; only contract data, so the old instance and its sources go meanwhile);
* `bound` with the object's variant and the members it follows.
* next attempt is announced with `loading`); `binding` while its object is live and its `bound` waits for a slot;
* `bound` with the object's variant and the members it follows. A `pending` hold unbound with `loading` and the
* `binding` after it, or a `binding` after a swap in place, keep the names it was bound with, `previous`: its consumer's
* calls sent by those names (before it learnt of the `loading`, or still bound) wait at its object's gate and are
* answered by name, design §22; only contract data, so the old instance and its sources go meanwhile.
*/
type HoldState =
| { readonly kind: "unplaced"; readonly request: HoldRequest }
| { readonly kind: "pending"; readonly toldFailed: boolean }
| { readonly kind: "pending"; readonly toldFailed: boolean; readonly previous?: BoundNames }
| { readonly kind: "binding"; readonly previous?: BoundNames }
| {
readonly kind: "bound";
@ -154,10 +155,19 @@ export class Hold implements Holder, UnboundRequest {
return this.#state.kind === "bound";
}
/** Whether calls on it reach its object: bound, or binding afresh after a swap in place (its `bound` not sent yet). */
/**
* Whether calls on it reach its object: bound, or by the names it was bound with while it rebinds (a swap in place, a
* `loading` unbind: its fresh `bound` not sent yet).
*/
get callable(): boolean {
return this.#names() !== undefined;
}
/** The names its calls are decoded by: its binding's, or the ones it was bound with while it rebinds. */
#names(): BoundNames | undefined {
const state = this.#state;
return state.kind === "bound" || (state.kind === "binding" && state.previous !== undefined);
if (state.kind === "bound") return state;
return state.kind === "pending" || state.kind === "binding" ? state.previous : undefined;
}
/** Whom it acts for (the unbound index's revocation). */
@ -280,21 +290,29 @@ export class Hold implements Holder, UnboundRequest {
}
/**
* Its object left `live` (a restart, a parked swap, a live failure): no longer subscribed or waiting for a slot, its
* calls in flight rejected (`error`, or `unavailable`), pending again; a `loading` tells the consumer it rebinds
* (a failure is sent by `#pending`, which a consumer applies to a bound hold as a rebinding too, model §5).
* Its object left `live` (a restart, a parked swap, a live failure): no longer subscribed or waiting for a slot,
* pending again. With `error` (a failure, sent by `#pending`, which a consumer applies to a bound hold as a rebinding
* too, model §5) its calls in flight are rejected. Without (`loading`, which tells the consumer it rebinds) they stay,
* and calls by the names it was bound with keep arriving (sent before its consumer learnt of the `loading`): each gets
* exactly one answer from what it waits on, the instance it already runs on (its result, or `released` at the seal)
* or its object's gate (the new instance's result, its failure, `park_timeout`, its end), design §22.
*/
#unbind(error: ChordError | undefined): void {
const previous = this.#names();
const state = this.#state;
if (state.kind === "bound") unfollow(state);
this.#changes = [];
this.#failed = [];
this.#retired = NO_RETIRED;
this.#touched = false;
const reason = error ?? new ChordError({ code: "unavailable", message: `hold ${this.id}: its object rebinds` });
this.#provider.unbound(this, reason);
this.#state = PENDING;
if (error === undefined) this.#provider.send({ kind: "loading", hold: this.id });
if (error !== undefined) {
this.#provider.unbound(this, error);
this.#state = PENDING;
return;
}
this.#provider.unbind(this);
this.#state = previous === undefined ? PENDING : { kind: "pending", toldFailed: false, previous };
this.#provider.send({ kind: "loading", hold: this.id });
}
/**
@ -331,7 +349,8 @@ export class Hold implements Holder, UnboundRequest {
if (current.kind !== "pending") return;
const state = object.state;
if (state.kind === "live") {
this.#state = BINDING;
// Calls by the names it was bound with are still answered until the fresh `bound` is sent.
this.#state = current.previous === undefined ? BINDING : { kind: "binding", previous: current.previous };
this.#provider.bind(this, this.#provider.link.context);
} else if (state.kind === "failed") {
// Each failure and each new cause (a parked object's blocker changed) is a message of its own.
@ -339,6 +358,7 @@ export class Hold implements Holder, UnboundRequest {
const retryIn = retryAt === undefined ? {} : { retryIn: Math.max(0, retryAt - Date.now()) };
const failed = { kind: cause.kind, error: wireError(cause.error) };
this.#provider.send({ kind: "failed", hold: this.id, cause: failed, ...retryIn });
// Told a failure, its consumer rejects what it still waits on: calls by the old names are not taken any more.
this.#state = { kind: "pending", toldFailed: true };
} else if (state.kind === "loading" && current.toldFailed) {
this.#provider.send({ kind: "loading", hold: this.id });
@ -359,13 +379,12 @@ export class Hold implements Holder, UnboundRequest {
}
/**
* The method at `index` of the bound variant and whether its result carries byte segments (its schema has a
* `bytes()` leaf). Throws on an index outside the variant (a protocol error); a hold that is not bound has none.
* The method at `index` of the bound variant (while it rebinds, of the names its consumer still has: its object answers
* by name) and whether its result carries byte segments (its schema has a `bytes()` leaf). Throws on an index outside
* the variant (a protocol error); a hold that is not callable has none.
*/
method(index: number): { readonly name: string; readonly resultBytes: boolean } {
const state = this.#state;
// Binding afresh after a swap: by the names the consumer still has (the new instance answers by name).
const bound = state.kind === "bound" ? state : state.kind === "binding" ? state.previous : undefined;
const bound = this.#names();
const name = bound?.names.methods[index];
if (name === undefined) throw new Error(`hold ${this.id}: no method ${index}`);
return { name, resultBytes: bound!.variant.methods.get(name)!.result.bytes };

17
packages/chord/src/remote/provider.ts
View file

@ -280,9 +280,9 @@ export class ProviderSide {
}
/**
* A served hold left its object's binding (slice 7b §6, model §5): its unsent pending ops and a queued `bound` are
* dropped (the ack slots of messages already sent stay until their `ack`), its calls in flight rejected with `error`
* (each slot freed once).
* A served hold left its object's binding with `error` (a failure, a removal or a version no longer bound; slice 7b
* §6, model §5): as `unbind`, and its calls in flight rejected with `error` (each slot freed once). A hold unbinding
* with `loading` keeps them (`unbind` alone).
*/
unbound(hold: Hold, error: ChordError): void {
this.unbind(hold);
@ -377,11 +377,12 @@ export class ProviderSide {
/**
* A call: one on a hold this side does not know is ignored (Q1), more than this side's N at once and compressed
* arguments from a child (clients never compress) are protocol errors. On a relayed hold it is forwarded, its
* arguments checked against this link's part limit first (the bytes are here). On a served hold that is not bound
* yet (nor binding afresh after a swap) it is refused `not_ready`; its method is an index into the bound variant (one
* outside it is a protocol error) and its arguments are opened and parsed here within the hold's limits (past one
* the call fails `too_large/part`; a malformed part is a protocol error), absent ones `null`. Arguments this host
* must inflate asynchronously are inflated first, then the call is handled again.
* arguments checked against this link's part limit first (the bytes are here). On a served hold that is not callable
* (not bound yet, nor rebinding by the names its consumer still has after a swap in place or a `loading`) it is
* refused `not_ready`; its method is an index into the bound variant (one outside it is a protocol error) and its
* arguments are opened and parsed here within the hold's limits (past one the call fails `too_large/part`; a
* malformed part is a protocol error), absent ones `null`. Arguments this host must inflate asynchronously are
* inflated first, then the call is handled again.
*/
#call(message: Extract<Message, { readonly kind: "call" }>): void | Promise<void> {
if (this.#calls.has(message.call)) throw new Error(`call ${message.call}: the id is in use`);

451
packages/chord/test/remote/hold.test.ts
View file

@ -9,7 +9,7 @@ import { createHost, defineFacet, type FacetHost, type Host, type Self } from ".
import type { ChordHost } from "../../src/host/internal.ts";
import { ChordObject } from "../../src/host/object.ts";
import { type Link, memoryLinkPair } from "../../src/link/index.ts";
import { binaryCodec, jsonPart } from "../../src/link/internal.ts";
import { binaryCodec, jsonPart, type Message } from "../../src/link/internal.ts";
import { Hold } from "../../src/remote/hold.ts";
import { constant, type Writable } from "../../src/state/index.ts";
import {
@ -32,6 +32,7 @@ import {
relayed,
SERVER_ADDRESS,
Server,
Session,
serveNotes,
statesOf,
type TestPrincipal,
@ -1500,7 +1501,7 @@ describe('calls over a link while a swap parks (decision "Swaps: dependents firs
await server.close(context);
});
it("a call already waiting at the gate when the swap parks is answered with its hold's loading, never not_ready", async () => {
it("a call already waiting at the gate when the swap parks survives its hold's loading: answered by the new code", async () => {
const store = await memoryStore();
const server = createHost<TestPrincipal>({
address: Server(),
@ -1566,11 +1567,17 @@ describe('calls over a link while a swap parks (decision "Swaps: dependents firs
{ object: "app.gate-provider", kind: "parked", cause: expect.objectContaining({ kind: "waiting" }) },
]);
await transported();
// Parked: its hold left the old instance and tells the client \`loading\`; the call that waited at the gate is
// answered as every call in flight on a hold that leaves its object (§21), never \`not_ready\` while live.
// Parked: its hold left the old instance and tells the client \`loading\`; the call that waited at the gate stays
// (decision "Calls in flight survive a \`loading\` unbind"), unanswered until the new attempt's outcome.
expect(states).toEqual(["loading"]);
expect(await outcome).toEqual({ code: "unavailable", message: "hold 0: its object rebinds" });
let answered = false;
void outcome.then(() => {
answered = true;
});
await transported();
expect(answered).toBe(false);
opened.resolve();
expect(await outcome).toEqual({ value: 2 });
await transported();
expect(states).toEqual(["loading", "live"]);
expect(await gated.bump(context)).toBe(2);
@ -1581,3 +1588,437 @@ describe('calls over a link while a swap parks (decision "Swaps: dependents firs
}
});
});
// A call sent after the hold's `loading` by the names it was bound with (as a consumer sends it before it learns of the
// `loading`) is kept as a call in flight at the unbind is: several tests below send their calls that way.
describe('calls in flight across a `loading` unbind (decision "Calls in flight survive a `loading` unbind")', () => {
/** A remote object on the server: `bump` (variant `a` only) and `ping` answer its code's number. */
const Kept = remote("app.kept", {
version: 1,
params: {},
hostAddress: () => Server(),
variants: {
a: { state: { label: str() }, methods: { bump: method(undefined, int()), ping: method(undefined, int()) } },
b: { state: { label: str() }, methods: { ping: method(undefined, int()) } },
},
});
const KEPT = wireNames(Kept, "a");
/** `Kept` v2: its label became an object, which v1 consumers are not mapped to (their version is dropped). */
const KeptV2 = evolve(Kept, (e) => e.state("a", "label", obj({ text: str() })));
/** A local type whose factory throws (with params: no host-lifetime object): one needing it parks on its failure. */
const Broken = local("app.kept-broken", { version: 1, params: { id: int() }, members: { n: int() } });
/** One code of `Kept`. */
type KeptCode = {
/** What `bump` and `ping` answer. */
readonly n: number;
readonly variant?: "a" | "b";
/** Whether it opens a native (a reload then restarts it), which opens once `opened` resolves. */
readonly native?: boolean;
readonly opened?: Promise<void>;
/** What `bump` waits for first. */
readonly gate?: Promise<void>;
/** Whether it needs `Broken` (it parks on a failed dependency). */
readonly broken?: boolean;
/** Each call that ran on this code, by method. */
readonly ran?: string[];
};
/** `Kept` as `code` (with `Broken` served beside it). */
function keptFacet(code: KeptCode) {
return defineFacet<TestPrincipal>((h) => {
h.serve(Broken, () => {
throw new Error("broken factory");
});
h.serve(Kept, (_, _principal, self) => {
if (code.native === true)
self.open(
async () => {
await code.opened;
return null;
},
() => {},
);
if (code.broken === true) self.get(Broken, { id: 1 });
const ping = async () => {
code.ran?.push("ping");
return code.n;
};
const bump = async () => {
await code.gate;
code.ran?.push("bump");
return code.n;
};
const label = constant(`v${code.n}`);
return code.variant === "b"
? { variant: "b", members: { label, ping } }
: { variant: "a", members: { label, bump, ping } };
});
});
}
/** A server serving `code` and a raw peer holding its `Kept` as hold 0, bound. */
async function keptServer(
code: KeptCode,
options: { readonly closeTimeout?: number; readonly readyTimeout?: number; readonly ack?: boolean } = {},
) {
const server = createHost<TestPrincipal>({
address: Server(),
principal: { kind: "server" },
closeTimeout: options.closeTimeout,
readyTimeout: options.readyTimeout,
revokedBy: (principal) => (principal.kind === "owner" ? ["owner"] : []),
onError: () => {},
});
await server.load([{ name: "@test/app", facet: keptFacet(code) }], context);
const [a, b] = memoryLinkPair();
server.accept(a, { principal: { kind: "owner" }, codec: testCodec() }, context);
const peer = new RawPeer(b, { ack: options.ack });
await peer.next("hello");
hold(peer, 0, Kept);
await peer.next("bound");
const reload = (next: KeptCode | ReturnType<typeof keptFacet>) =>
server.reload([{ name: "@test/app", facet: "n" in next ? keptFacet(next) : next }], context);
return { server, peer, reload };
}
/** Sends call `call` of `method` (an index of `Kept`'s variant `a`) on hold 0. */
function call(peer: RawPeer, call: number, method: "bump" | "ping"): void {
peer.send({ kind: "call", call, hold: 0, method: KEPT.methods[method]! });
}
/** The answers (`result`, `error`) the peer received for `call`, taken. */
function answersTo(peer: RawPeer, id: number): Message[] {
const answers = peer.received.filter(
(message) => (message.kind === "result" || message.kind === "error") && message.call === id,
);
peer.received.splice(0, peer.received.length, ...peer.received.filter((message) => !answers.includes(message)));
return answers;
}
it("a restart: the running call drains on the old code, one arriving during the drain runs on the new code", async () => {
const gate = Promise.withResolvers<void>();
const opened = Promise.withResolvers<void>();
const ran: string[] = [];
const { server, peer, reload } = await keptServer({ n: 1, native: true, gate: gate.promise, ran });
call(peer, 0, "bump");
await transported();
const report = await reload({ n: 2, native: true, opened: opened.promise, ran });
expect(report.objects).toEqual([{ object: "app.kept", kind: "restarted" }]);
// The hold stays bound while the old instance drains: a call sent now waits at the gate.
call(peer, 1, "bump");
await transported();
gate.resolve();
expect(partValue((await peer.next("result")).value!)).toBe(1);
await peer.next("loading");
await transported();
// Kept across the `loading`: no answer until the new attempt is live.
expect(answersTo(peer, 1)).toEqual([]);
opened.resolve();
await peer.next("bound");
const result = await peer.next("result");
expect([result.call, partValue(result.value!)]).toEqual([1, 2]);
expect(ran).toEqual(["bump", "bump"]);
expect(peer.kinds().filter((kind) => kind === "error")).toEqual([]);
await server.close(context);
});
it("a call stuck on the old code past closeTimeout is answered released at the seal, once", async () => {
const opened = Promise.withResolvers<void>();
const { server, peer, reload } = await keptServer(
{ n: 1, native: true, gate: new Promise(() => {}) },
{ closeTimeout: 30 },
);
call(peer, 0, "bump");
await transported();
await reload({ n: 2, native: true, opened: opened.promise });
await peer.next("loading");
const error = await peer.next("error");
expect(error).toMatchObject({ call: 0, error: { error: { code: "released" } } });
opened.resolve();
await peer.next("bound");
await transported();
expect(answersTo(peer, 0)).toEqual([]);
await server.close(context);
});
it("a kept call takes the gate's rejection: the parked cause, or unavailable after readyTimeout", async () => {
const parked = await keptServer({ n: 1, native: true });
const opened = Promise.withResolvers<void>();
await parked.reload({ n: 2, native: true, opened: opened.promise, broken: true });
call(parked.peer, 0, "ping");
opened.resolve();
const failed = await parked.peer.next("failed");
expect(failed.cause).toMatchObject({ kind: "waiting" });
const rejected = await parked.peer.next("error");
expect(rejected).toMatchObject({
call: 0,
error: { error: { message: expect.stringContaining("broken factory") } },
});
await parked.server.close(context);
const timed = await keptServer({ n: 1, native: true }, { readyTimeout: 60 });
await timed.reload({ n: 2, native: true, opened: new Promise(() => {}) });
call(timed.peer, 0, "ping");
await timed.peer.next("loading");
const timeout = await timed.peer.next("error");
// Bounded by readyTimeout (here the new attempt's own deadline fails it first).
expect(timeout).toMatchObject({
call: 0,
error: { error: { code: "unavailable", message: expect.stringContaining("60 ms") } },
});
await timed.server.close(context);
});
it("a kept call aborted or released by its consumer gets no answer and never runs", async () => {
const ran: string[] = [];
const opened = Promise.withResolvers<void>();
const { server, peer, reload } = await keptServer({ n: 1, native: true, ran });
await reload({ n: 2, native: true, opened: opened.promise, ran });
call(peer, 0, "ping");
call(peer, 1, "ping");
await peer.next("loading");
peer.send({ kind: "abort", call: 0 });
await transported();
opened.resolve();
await peer.next("bound");
expect((await peer.next("result")).call).toBe(1);
await transported();
expect(answersTo(peer, 0)).toEqual([]);
expect(ran).toEqual(["ping"]);
// A release while kept: the call goes with the hold.
const second = Promise.withResolvers<void>();
await reload({ n: 3, native: true, opened: second.promise, ran });
call(peer, 2, "ping");
await peer.next("loading");
peer.send({ kind: "release", hold: 0 });
await transported();
second.resolve();
await transported();
expect(answersTo(peer, 2)).toEqual([]);
expect(ran).toEqual(["ping"]);
await server.close(context);
});
it("across a variant change a kept call runs by name: one the new variant lacks fails invalid/member", async () => {
const opened = Promise.withResolvers<void>();
const { server, peer, reload } = await keptServer({ n: 1, native: true });
await reload({ n: 2, variant: "b", native: true, opened: opened.promise });
call(peer, 0, "bump");
call(peer, 1, "ping");
await peer.next("loading");
opened.resolve();
const fresh = await peer.next("bound");
expect(fresh.holds[0]!.variant).toBe(wireNames(Kept, "b").variant);
const missing = await peer.next("error");
expect(missing).toMatchObject({ call: 0, error: { error: { code: "invalid", reason: "member" } } });
const result = await peer.next("result");
expect([result.call, partValue(result.value!)]).toEqual([1, 2]);
await server.close(context);
});
it("a kept call of a version the new code drops is rejected contract_mismatch with the hold's failed", async () => {
const opened = Promise.withResolvers<void>();
const ran: string[] = [];
const { server, peer, reload } = await keptServer({ n: 1, native: true });
const v2 = defineFacet<TestPrincipal>((h) =>
h.serve(KeptV2, (_, _principal, self) => {
self.open(
async () => {
await opened.promise;
return null;
},
() => {},
);
const ping = async () => {
ran.push("ping");
return 2;
};
return { variant: "a", members: { label: constant({ text: "v2" }), bump: ping, ping } };
}),
);
await reload(v2);
call(peer, 0, "ping");
await peer.next("loading");
opened.resolve();
const failed = await peer.next("failed");
expect(failed.cause).toMatchObject({ kind: "waiting", error: { error: { code: "contract_mismatch" } } });
expect(await peer.next("error")).toMatchObject({ call: 0, error: { error: { code: "contract_mismatch" } } });
expect(ran).toEqual([]);
await server.close(context);
});
it("a call by the old names crossing the loading on the wire waits at the gate, never not_ready", async () => {
const opened = Promise.withResolvers<void>();
const { server, peer, reload } = await keptServer({ n: 1, native: true });
await reload({ n: 2, native: true, opened: opened.promise });
// Sent as a consumer sends it before it learns of the `loading`: by the names it was bound with.
await peer.next("loading");
call(peer, 0, "bump");
await transported();
expect(answersTo(peer, 0)).toEqual([]);
opened.resolve();
await peer.next("bound");
const result = await peer.next("result");
expect([result.call, partValue(result.value!)]).toEqual([0, 2]);
await server.close(context);
});
it("binding with the previous names, then a restart: the call running on the swapped-in instance drains to its result", async () => {
const gate = Promise.withResolvers<void>();
const opened = Promise.withResolvers<void>();
const { server, peer, reload } = await keptServer({ n: 1 }, { ack: false });
// A second `bound` in flight fills the window (two); the next swap in place leaves its fresh `bound` waiting.
await reload({ n: 2 });
await peer.next("bound");
await reload({ n: 3, gate: gate.promise });
await transported();
expect(peer.kinds()).toEqual([]);
call(peer, 0, "bump");
await transported();
// A restart unbinds the binding hold at once: the call runs on the live (unsealed) instance and stays.
await reload({ n: 4, native: true, opened: opened.promise });
await peer.next("loading");
await transported();
expect(answersTo(peer, 0)).toEqual([]);
gate.resolve();
const result = await peer.next("result");
expect([result.call, partValue(result.value!)]).toEqual([0, 3]);
opened.resolve();
peer.ack();
peer.ack();
await peer.next("bound");
await server.close(context);
});
it("a revocation racing the rebind ends the kept call with the link: no result", async () => {
const ran: string[] = [];
const { server, peer, reload } = await keptServer({ n: 1, native: true, ran });
await reload({ n: 2, native: true, opened: new Promise(() => {}), ran });
call(peer, 0, "ping");
await peer.next("loading");
server.revoke("owner", context);
await transported();
expect(peer.closed).toBe(true);
expect(answersTo(peer, 0).filter(({ kind }) => kind === "result")).toEqual([]);
expect(ran).toEqual([]);
await server.close(context);
});
it("a failed unbind still rejects: a live failure answers its calls with the failure, later ones not_ready", async () => {
const gate = Promise.withResolvers<void>();
let self: Self<TestPrincipal> | undefined;
const server = createHost<TestPrincipal>({ address: Server(), principal: { kind: "server" }, onError: () => {} });
const facet = defineFacet<TestPrincipal>((h) =>
h.serve(Kept, (_, _principal, declared) => {
self = declared;
const ping = async () => {
await gate.promise;
return 1;
};
return { variant: "a", members: { label: constant("v1"), bump: ping, ping } };
}),
);
await server.load([{ name: "@test/app", facet }], context);
const objects = instancesOf(ChordObject, "attempt");
const [a, b] = memoryLinkPair();
server.accept(a, { principal: { kind: "owner" }, codec: testCodec() }, context);
const peer = new RawPeer(b);
await peer.next("hello");
hold(peer, 0, Kept);
await peer.next("bound");
call(peer, 0, "ping");
await transported();
failLive(objects, self!, new ChordError({ code: "unavailable", message: "it broke" }));
const rejected = await peer.next("error");
expect(rejected).toMatchObject({ call: 0, error: { error: { message: "it broke" } } });
await peer.next("failed");
// Told a failure, the hold takes no calls by the old names.
call(peer, 1, "ping");
expect(await peer.next("error")).toMatchObject({ call: 1, error: { error: { code: "not_ready" } } });
gate.resolve();
await server.close(context);
});
describe("through a relay", () => {
/** `Kept` placed on a session's worker. */
const PlacedKept = remote("app.placed-kept", {
version: 1,
params: { session: str() },
hostAddress: ({ session }) => Session({ id: session }),
variants: { a: { state: { label: str() }, methods: { ping: method(undefined, int()) } } },
});
/** Serves `PlacedKept` answering `n` (recorded in `ran`), with a native that opens once `opened` resolves. */
function servePlacedKept(h: FacetHost<TestPrincipal>, n: number, ran: number[], opened?: Promise<void>): void {
h.serve(PlacedKept, (_, _principal, self) => {
self.open(
async () => {
await opened;
return null;
},
() => {},
);
const ping = async () => {
ran.push(n);
return n;
};
return { variant: "a", members: { label: constant(`v${n}`), ping } };
});
}
it("a worker's restart: the tab's kept call crosses the forwarded loading and is answered by the new code", async () => {
const ran: number[] = [];
const { tab, workers, server } = await relayed({ worker: (h) => servePlacedKept(h, 1, ran) });
const kept = await tab.get(PlacedKept, { session: "S1" }, context);
const states = statesOf(await kept.state.watch(context));
const opened = Promise.withResolvers<void>();
const worker = workers.get("session|S1")!;
await worker.reload(
[
{
name: "@test/worker",
facet: defineFacet<TestPrincipal>((h) => servePlacedKept(h, 2, ran, opened.promise)),
},
],
context,
);
// Still live during the drain: the call goes out and waits at the worker's gate.
const answer = kept.ping(context);
await transported();
expect(states).toEqual(["loading"]);
opened.resolve();
expect(await answer).toBe(2);
await transported();
expect(states).toEqual(["loading", "live"]);
expect(ran).toEqual([2]);
await server.close(context);
});
it("a downstream drop while the call is kept aborts it on the worker: it never runs", async () => {
const ran: number[] = [];
const { tab, workers, server, tabLink } = await relayed({ worker: (h) => servePlacedKept(h, 1, ran) });
const kept = await tab.get(PlacedKept, { session: "S1" }, context);
const opened = Promise.withResolvers<void>();
const worker = workers.get("session|S1")!;
await worker.reload(
[
{
name: "@test/worker",
facet: defineFacet<TestPrincipal>((h) => servePlacedKept(h, 2, ran, opened.promise)),
},
],
context,
);
const answer = rejection(kept.ping(context));
await transported();
tabLink.close();
expect(await answer).toMatchObject({ code: "unavailable" });
await transported();
opened.resolve();
await transported();
expect(ran).toEqual([]);
await server.close(context);
});
});
});