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

fix(chord): Chord v2 follows its design where the audit of slices 1-6a found it did not

Browse source 
A failed activation releases its references after its natives closed; a dependency going live while
the dependent activates retries it at once; a call on a local object runs with the host's principal
and incarnation; retiring an unloaded document checks its stored version and fingerprint, and a cut
retirement no longer fails its successor; ref.retry() re-runs a refused hold's start on the same id
(the provider keeps refused holds), and ends released/closed on a closed host; a relayed hold waits
while the parent link is down; a link drop while a dependency is still loading is no failure.
This commit is contained in:
Mario Zechner 2026-10-28 19:50:41 +00:00
parent 92c1da1f62
commit 391e83fb17
19 changed files with 514 additions and 104 deletions
Download patch file Download diff file Expand all files Collapse all files

41
packages/chord/src/next/documents/persistence.ts
View file

@ -29,6 +29,9 @@ export type Stored = {
readonly deltasSinceBase: number;
};
/** A live document found without loading it: what is stored of it and the fingerprint it was stored with. */
export type FoundDocument = { readonly stored: Stored; readonly fingerprint: string };
/** A loaded document: its value at the definition's version, and what is stored of it. */
export type Loaded = { readonly value: JsonObject; readonly stored: Stored };
@ -84,28 +87,27 @@ export class Persistence {
*/
decode(document: StoredDocument, definition: AnyDefinition, context: Context): Loaded {
const { record, deltasSinceBase } = document;
const { name, version, fingerprint } = definition;
if (record.version > version) {
throw new Error(`${name}: stored at version ${record.version}, newer than the code's ${version}`);
}
if (record.version === version && record.fingerprint !== fingerprint) {
throw new Error(`${name}: stored with another schema at version ${version} (a schema changed without a bump)`);
}
checkReadable(record, definition);
// The engine stores what commits gave it: JSON objects of the record's version.
const stored = document.value as JsonObject;
const value = record.version === version ? stored : definition.migrate(stored, record.version, context);
const value =
record.version === definition.version ? stored : definition.migrate(stored, record.version, context);
return {
value,
stored: { id: record.id, revision: record.revision, version: record.version, deltasSinceBase },
};
}
/** What is stored of the live document at `key` without loading it (a retirement's find), `undefined` when absent. */
async find(key: string, context: Context): Promise<Stored | undefined> {
/**
* What is stored of the live document at `key` without loading it (a retirement's find) and the schema fingerprint
* it was stored with (`checkReadable` takes both), `undefined` when absent.
*/
async find(key: string, context: Context): Promise<FoundDocument | undefined> {
const record = await this.store.read((executor) => findDocument(executor, this.at(key)), context);
if (record === undefined) return undefined;
// The change counter is all a retirement needs; `deltasSinceBase` is not read for one.
return { id: record.id, revision: record.revision, version: record.version, deltasSinceBase: 0 };
const stored = { id: record.id, revision: record.revision, version: record.version, deltasSinceBase: 0 };
return { stored, fingerprint: record.fingerprint };
}
/** Up to `limit` keys of live documents after `after`, in code point order. */
@ -161,6 +163,23 @@ export class Persistence {
}
}
/**
* Throws when `definition` cannot read a document stored at `stored`'s version and fingerprint: a newer version than
* the code's, or the same version with another fingerprint (D9). A load and a retirement check alike (R15).
*/
export function checkReadable(
stored: { readonly version: number; readonly fingerprint: string },
definition: AnyDefinition,
): void {
const { name, version, fingerprint } = definition;
if (stored.version > version) {
throw new Error(`${name}: stored at version ${stored.version}, newer than the code's ${version}`);
}
if (stored.version === version && stored.fingerprint !== fingerprint) {
throw new Error(`${name}: stored with another schema at version ${version} (a schema changed without a bump)`);
}
}
/** What is stored after a commit's `change` was adopted. */
export function storedAfter(change: StoredChange): Stored | undefined {
switch (change.kind) {

25
packages/chord/src/next/documents/transaction.ts
View file

@ -4,6 +4,7 @@
import { type Context, withoutAbortSignal } from "../../context/index.ts";
import { type Change, type Draft, type Prepared, type Tracker, track } from "../../delta/index.ts";
import { ChordError } from "../data/index.ts";
import { asChordError } from "../data/internal.ts";
import type { JobHandle, Outcome, Run, StartOptions } from "../jobs/index.ts";
import type { Job } from "../jobs/internal.ts";
import type { Columns, PutRowOf, RowOf } from "../store/columns.ts";
@ -20,8 +21,8 @@ import type { Documents } from "./documents.ts";
import type { SealToken } from "./fence.ts";
import type { Keyed, KeyedHandle, Singleton, SingletonHandle } from "./handles.ts";
import type { KeyedSlot } from "./keyed-slot.ts";
import type { Stored, StoredChange } from "./persistence.ts";
import type { DocumentSlot, SingletonSlot } from "./slot.ts";
import { checkReadable, type FoundDocument, type Stored, type StoredChange } from "./persistence.ts";
import type { DocumentSlot, PersistedSlot, SingletonSlot } from "./slot.ts";
/** What `self.tx` gives its callback. Inside it: only `await tx.*`, no other `self.tx`, no calls to other objects. */
export interface Tx {
@ -213,9 +214,14 @@ export class Transaction implements Tx {
const at = key === "" ? key : slot.key(key);
const entry = this.retireSlot(slot, at);
const persisted = slot.persisted;
// A cold persisted document is found, not loaded (one read; absent: a no-op, R13).
// A cold persisted document is found, not loaded (one read; absent: a no-op, R13); a version or fingerprint
// the code cannot read fails the owner as a load would (R15): a retirement of it is no exception.
if (entry.found === undefined && persisted?.cold(at)) {
entry.found = await this.#track(persisted.persistence.find(at, this.#context));
const found = await this.#track(persisted.persistence.find(at, this.#context));
// Cut meanwhile: its attempt was sealed, and failing the owner now would fail its successor.
this.#assertOpen();
if (found !== undefined) checkFound(found, persisted, slot);
entry.found = found?.stored;
}
}
@ -533,3 +539,14 @@ export class Transaction implements Tx {
return entry.draft.change.state;
}
}
/** Throws, failing `slot`'s owner `broken`, when its definition cannot read the cold document `found` (R15). */
function checkFound(found: FoundDocument, persisted: PersistedSlot, slot: DocumentSlot): void {
try {
checkReadable({ version: found.stored.version, fingerprint: found.fingerprint }, persisted.definition.current);
} catch (error) {
const failure = asChordError(error, slot.owner.label);
slot.owner.fail(failure);
throw failure;
}
}

7
packages/chord/src/next/host/attempt.ts
View file

@ -408,9 +408,10 @@ export class Attempt implements StoreSelf, KeyDeclarer {
const checked = checkArgs(served.provider, method, args, served.contract.version, context);
const linked = withLinkedAbortSignal(this.#abort.signal, context);
const callContext = linked.context;
// A local object's handlers act as the host; a remote one's as their caller.
const principal = served.kind === "remote" ? caller.principal : host.principal;
const ctx = new CallContext(callContext, { principal, incarnation: caller.incarnation }, host);
// A local object's handlers act as the host, its principal and its incarnation (§4); a remote one's as their
// caller. The caller stays the reference's (presence, §19).
const acting = served.kind === "remote" ? caller : { principal: host.principal, incarnation: host.incarnation };
const ctx = new CallContext(callContext, acting, host);
// Admitted to the drain before the handler runs: the handler may end its own object before it returns.
const running = Promise.withResolvers<unknown>();
host.count("calls", 1);

20
packages/chord/src/next/host/host.ts
View file

@ -181,8 +181,9 @@ export interface Host<P extends Principal = Principal, R extends JsonValue = Jso
get<C extends Contract>(Type: C, params: ParamsOf<C>, context: Context, options?: GetOptions): Promise<Reference<C>>;
/**
* A reference at once, following its object until released: loading, live, failed (retried as its object is) and
* loading again while its link is down; it never rejects. Throws at once for invalid params and a local type not
* served here; before `load` resolved these are decided once it did, failing the reference broken. Its `variant` is `undefined` until its first binding.
* loading again while its link is down; it never rejects. Throws at once for invalid params (also before `load`)
* and a local type not served here; before `load` resolved, a type not served here fails the reference broken once
* it did. Its `variant` is `undefined` until its first binding.
*/
hold<C extends Contract>(Type: C, params: ParamsOf<C>, context: Context, options?: GetOptions): Held<C>;
/**
@ -641,9 +642,9 @@ export class ChordHost<P extends Principal = Principal, R extends JsonValue = Js
}
/**
* A reference to `Type` made by `caller` for `host.hold` and getters: params and where it goes checked at once
* (`invalid/params`, `#assertRoute`), then bound (a placement or route that fails fails the reference broken);
* before `load` resolved, both once it did (a host closed meanwhile releases it).
* A reference to `Type` made by `caller` for `host.hold` and getters: params checked at once (`invalid/params`,
* also before `load`), where it goes too (`#assertRoute`), then bound (a placement or route that fails fails the
* reference broken); before `load` resolved, where it goes once it did (a host closed meanwhile releases it).
*/
#hold(Type: Contract, params: unknown, caller: Caller, context: Context, options: GetOptions): Binding {
this.#assertOpen();
@ -667,14 +668,19 @@ export class ChordHost<P extends Principal = Principal, R extends JsonValue = Js
/**
* Binds a reference (`bind`); where it cannot go (`#assertRoute`, a placement or `authorize` that throws or refuses)
* fails it broken.
* fails it broken, until `ref.retry()` binds it again.
*/
#bindOrFail(binding: Binding, Type: Contract, checked: CheckedParams, context: Context): void {
try {
this.#assertRoute(Type);
this.bind(binding, Type, checked, context);
} catch (error) {
binding.refuse({ kind: "broken", error: asChordError(error, Type.name) });
// A retry after the host closed ends the reference as the close did the others.
const again = () =>
this.#state === "closed"
? binding.end(this.#closedError())
: this.#bindOrFail(binding, Type, checked, context);
binding.refuse({ kind: "broken", error: asChordError(error, Type.name) }, again);
}
}

22
packages/chord/src/next/host/object.ts
View file

@ -125,6 +125,8 @@ export class ChordObject implements Target {
readonly #holders = new Set<Holder>();
/** Objects this object's activation waits for; followed to find cycles (Q5). */
readonly #waitingFor = new Set<Target>();
/** The attempt a dependency went live during (R18): if it fails on a dependency, it is retried at once. */
#woken: Attempt | undefined;
/** The current attempt's outcome: resolves when it goes live, rejects with its error; renewed by every retry. */
#outcome = settlement();
/** The error of the last failed attempt, until one goes live. */
@ -474,9 +476,17 @@ export class ChordObject implements Target {
}
}
/** A dependency went live: retried at once (backoff window skipped) when it failed on a dependency (R18). */
/**
* A dependency went live: retried at once (backoff window skipped) when it failed on a dependency (R18). A wake
* while it activates is remembered: that activation may still fail on the dependency's earlier failure.
*/
dependencyLive(): void {
if (this.#state.kind !== "failed" || this.#state.holding === undefined) return;
const state = this.#state;
if (state.kind === "activating") {
this.#woken = state.attempt;
return;
}
if (state.kind !== "failed" || state.holding === undefined) return;
this.host.backoff.skip(this.identity);
this.attempt();
}
@ -519,6 +529,7 @@ export class ChordObject implements Target {
}
#wentLive(attempt: Attempt): void {
this.#woken = undefined;
this.#state = { kind: "live", attempt };
this.#exposed = attempt;
if (this.#members.size > 0 || !this.#keys.empty)
@ -538,6 +549,8 @@ export class ChordObject implements Target {
* broken: no automatic retry.
*/
#failed(attempt: Attempt, cause: FailureCause, holding: Attempt | undefined): void {
const woken = this.#woken === attempt;
this.#woken = undefined;
// What its activation loaded for the pinned keys goes; they stay frozen.
this.#keys.discard();
if (cause.kind === "transient") this.host.backoff.failed(this.identity);
@ -546,9 +559,10 @@ export class ChordObject implements Target {
this.#state = { kind: "failed", failure, holding };
this.#lastFailure = cause.error;
attempt.seal(this.ended());
if (holding === undefined) attempt.releaseReferences();
const closing = attempt.close(END).then((failures) => {
attempt.detach();
// Released once its natives closed (a close may use them, §6, §13); R18's attempt keeps them for the retry.
if (holding === undefined) attempt.releaseReferences();
return failures;
});
this.host.ending(this.identity, this.#reported(closing));
@ -557,6 +571,8 @@ export class ChordObject implements Target {
this.#notify();
// Job runs kept it while it activated; failed, they keep it no longer.
this.#idle();
// A dependency went live while this attempt activated: the wake it missed retries it now (R18).
if (holding !== undefined && woken) this.dependencyLive();
}
/** The member source of `member`, made bound to `exposed`'s member by its first use. */

6
packages/chord/src/next/host/placement.ts
View file

@ -51,15 +51,15 @@ export class Placement {
/**
* Where a hold arriving on `link` acting as `principal` goes: here when its type is served here; over the route of
* the address it names; else over the parent link, unless it came on the parent link or there is none
* (`unavailable`). Throws as `reference` does.
* the address it names; else over the parent link (waiting for it while it is down), unless it came on the parent
* link or this host never had one (`unavailable`). Throws as `reference` does.
*/
hold(request: HoldRequest, link: HostLink, principal: Principal | undefined, context: Context): Placed {
if (this.#types.has(request.type)) return HERE;
const { address } = request;
const routed = this.#pathTo(address, principal, context);
if (routed !== undefined) return { kind: "path", path: routed, address };
if (link.role === "parent" || this.#parent.link === undefined) {
if (link.role === "parent" || !this.#parent.linked) {
throw new ChordError({ code: "unavailable", message: `${request.type} is not served here` });
}
return { kind: "path", path: this.#parent, address };

24
packages/chord/src/next/host/reference.ts
View file

@ -123,7 +123,7 @@ type BindingState =
readonly dependent: ChordObject | undefined;
readonly target: Target;
}
| { readonly kind: "refused"; readonly batch: Batch; readonly failure: Failed }
| { readonly kind: "refused"; readonly batch: Batch; readonly failure: Failed; readonly bind: () => void }
| { readonly kind: "released"; readonly state: Ended };
/** What a wait for a state decides at each change. */
@ -240,11 +240,15 @@ export class Binding implements Holder {
this.moved();
}
/** A `host.hold` that cannot be bound once `load` resolved: failed with `cause`, which no retry changes. */
refuse(cause: FailureCause): void {
/**
* A `host.hold` that cannot be bound (where it goes refused it): failed with `cause` until `ref.retry()` runs `bind`
* again (a route's `authorize` may admit it later).
*/
refuse(cause: FailureCause, bind: () => void): void {
const state = this.#state;
if (state.kind !== "unbound") return;
this.#state = { kind: "refused", batch: state.batch, failure: { kind: "failed", cause, retryAt: undefined } };
const failure: Failed = { kind: "failed", cause, retryAt: undefined };
this.#state = { kind: "refused", batch: state.batch, failure, bind };
this.moved();
}
@ -388,9 +392,17 @@ export class Binding implements Holder {
return this.#ended.promise;
}
/** `ref.retry()`: the object is asked for an attempt now; it ignores this unless failed. */
/** `ref.retry()`: the object is asked for an attempt now (ignored unless failed); a refused hold is bound again. */
retry(): void {
if (this.#state.kind === "bound") this.#state.target.retry();
const state = this.#state;
if (state.kind === "bound") {
state.target.retry();
return;
}
if (state.kind !== "refused") return;
this.#state = { kind: "unbound", batch: state.batch, dependent: undefined };
this.moved();
state.bind();
}
/** Releases the reference (idempotent); the object may end when it was its last. */

4
packages/chord/src/next/jobs/persisted.ts
View file

@ -69,8 +69,8 @@ export class PersistedRuns {
/** Finds what is stored of run `id`'s state document (a run of another job version), keeping it for the drop. */
async findBlockedState(id: string, context: Context): Promise<void> {
const stored = await this.#states.find(id, context);
if (stored !== undefined) this.#blockedStates.set(id, stored);
const found = await this.#states.find(id, context);
if (found !== undefined) this.#blockedStates.set(id, found.stored);
}
/** Every stored record of the job, in run id order, page by page with plain reads (activation holds the line). */

34
packages/chord/src/next/remote/hold.ts
View file

@ -72,6 +72,8 @@ type HoldState =
/** The members it follows, in the order they were subscribed. */
readonly followed: Followed[];
}
/** Refused where it arrived (placement, `authorize`, a mismatch, invalid params, a cycle): `retry` starts it again. */
| { readonly kind: "refused"; readonly request: HoldRequest }
| { readonly kind: "released" };
/** A bound hold's state. */
@ -84,8 +86,8 @@ const BINDING: HoldState = { kind: "binding" };
/**
* One hold served here: a reference to its object acting as its caller (the hold's principal). While pending, each
* failure of its object is sent as `failed` (with the rest of its backoff window) and the next attempt as `loading`; a
* hold that cannot be taken (a refused type, a mismatch, invalid params or placement) is a broken `failed` that ends
* it. Once live it binds when its link's window has a slot, then sends a part per batch that changed its members
* hold that cannot be taken (a refused type, a mismatch, invalid params or placement) is a broken `failed` that holds
* nothing until a `retry` starts it again. Once live it binds when its link's window has a slot, then sends a part per batch that changed its members
* (or keeps their ops pending while the window is full), and `gone` when its object ends.
*/
export class Hold implements Holder {
@ -137,7 +139,7 @@ export class Hold implements Holder {
* Admits the hold once `load` resolved: its params within the link's part limit (`too_large/part`), then relayed
* (it is replaced by a relayed hold), or served here, its params opened and parsed (a malformed part is a protocol
* error), its object taken and followed (`moved`). A hold that cannot go anywhere, a type that does not match,
* invalid params, a wrong placement or a cycle (Q5) is a broken `failed`, which ends it.
* invalid params, a wrong placement or a cycle (Q5) is a broken `failed`; the hold is kept, refused, for a `retry`.
*/
async start(request: HoldRequest): Promise<void> {
const { link } = this.#provider;
@ -169,7 +171,7 @@ export class Hold implements Holder {
}
object = link.host.serving(request, params, this.caller.principal!, link.context);
} catch (error) {
return this.#refuse(asChordError(error, request.type));
return this.#refuse(request, asChordError(error, request.type));
}
// Counted before the object lists it in its presence: a derive over that may close this link synchronously.
this.#object = object;
@ -203,9 +205,19 @@ export class Hold implements Holder {
}
}
/** A `retry` message: the object is asked for an attempt now (it ignores this unless failed). */
/**
* A `retry` message: the object is asked for an attempt now (it ignores this unless failed); a refused hold is
* started again (a route's `authorize` may admit it now, a type may be served here now).
*/
retry(): void {
this.#object?.retry();
const state = this.#state;
if (state.kind !== "refused") {
this.#object?.retry();
return;
}
this.#state = PENDING;
this.#provider.send({ kind: "loading", hold: this.id });
void this.start(state.request);
}
/** Calls a method of the held object as the hold's caller. */
@ -413,11 +425,13 @@ export class Hold implements Holder {
this.#provider.touched(this);
}
/** Refuses the hold: a broken `failed`, final; the provider forgets it. */
#refuse(error: ChordError): void {
this.#provider.forget(this);
/** Refuses the hold: a broken `failed`; it keeps `request` for a `retry`, holding nothing until then. */
#refuse(request: HoldRequest, error: ChordError): void {
const { params } = request;
// Its params owned, not a view of the message they came in (which the link's buffers may hold).
const owned = typeof params === "string" ? params : (params as Uint8Array).slice();
this.#state = { kind: "refused", request: { ...request, params: owned } };
this.#provider.send({ kind: "failed", hold: this.id, cause: { kind: "broken", error: wireError(error) } });
this.release();
}
}

8
packages/chord/src/next/remote/path.ts
View file

@ -18,6 +18,8 @@ export class Path {
readonly #holders = new Set<Upstream>();
readonly #changed: () => void;
#link: HostLink | undefined;
/** Whether it ever had a link: a host never connected to a parent has no parent path to wait on. */
#linked = false;
constructor(changed: () => void) {
this.#changed = changed;
@ -28,6 +30,11 @@ export class Path {
return this.#link;
}
/** Whether it had a link once (it may be down now: its holders wait for the next). */
get linked(): boolean {
return this.#linked;
}
/** Whether anything is held over this path. */
get held(): boolean {
return this.#holders.size > 0;
@ -49,6 +56,7 @@ export class Path {
/** `link` is the path's link now: every holder is held on it. */
connected(link: HostLink): void {
this.#link = link;
this.#linked = true;
for (const holder of this.#holders) holder.attach(link);
}

42
packages/chord/src/next/remote/remote-object.ts
View file

@ -153,10 +153,10 @@ export class RemoteObject implements Target {
this.#hold = { side: link.consumer, id: link.consumer.attach(this, chain) };
}
/** The link dropped with `error`: loading until a new link's hold is bound (`#startOver`). */
dropped(error: ChordError): void {
/** The link dropped: loading until a new link's hold is bound (`#startOver`). */
dropped(_error: ChordError): void {
this.#hold = undefined;
this.#startOver(error);
this.#startOver();
}
/**
@ -214,16 +214,21 @@ export class RemoteObject implements Target {
}
/**
* `ref.retry()`: a failed object's provider is asked for an attempt; one that failed for a part past the link's
* limits is placed anew and held with a new id, now or on its path's next link (a placement that throws fails it
* broken again); ignored otherwise and without a link.
* `ref.retry()`: a failed object's provider is asked for an attempt (a hold it refused is started there again). One
* that failed for a part past the link's limits is placed anew and held with a new id, now or on its path's next
* link (a placement that throws fails it broken again).
*/
retry(): void {
if (this.#state.kind !== "failed") return;
if (this.#path !== this.#host.unplaced) {
this.#hold?.side.retry(this.#hold.id);
if (this.#path === this.#host.unplaced) {
this.#place();
return;
}
this.#hold?.side.retry(this.#hold.id);
}
/** Places a reference whose part was past the limits anew, on its path now (a placement that throws fails it). */
#place(): void {
let path: Path;
try {
path = this.#host.pathOf(this.contract, this.#params, this.caller.principal);
@ -334,9 +339,7 @@ export class RemoteObject implements Target {
* (`#startOver`).
*/
loading(): void {
this.#startOver(
new ChordError({ code: "unavailable", message: `${this.label}: its relay's upstream link dropped` }),
);
this.#startOver();
}
/**
@ -402,18 +405,13 @@ export class RemoteObject implements Target {
}
/**
* Loading again until the next binding; the members keep their last values. A dependent's activation waiting for
* this hold while it was loading already fails at once with `error` (transient) and is retried when a new hold is
* bound (R18).
* Loading again until the next binding; the members keep their last values. A drop while it is loading already
* changes nothing: a dependent's activation waiting for it goes on waiting for the next link's binding, bounded by
* its `readyTimeout` (§6, R18; a drop is not a dependency failure).
*/
#startOver(error: ChordError): void {
if (this.#state.kind === "ended") return;
if (this.#state === LOADING) {
this.#settled.reject(error);
this.#settled = settlement();
} else {
this.#move(LOADING);
}
#startOver(): void {
const { kind } = this.#state;
if (kind !== "ended" && kind !== "loading") this.#move(LOADING);
}
/** Moves to `state` and tells the reference; `settled` follows (a new wait each time it is loading again). */

90
packages/chord/test/next/documents/keyed-slot.test.ts
View file

@ -449,6 +449,96 @@ describe("cold keyed loads on the line", () => {
await host.close(context);
});
it.each([
["a newer version", 2, { title: str(), n: str() }, "stored at version 2, newer than the code's 1"],
["another fingerprint at its version", 1, { title: str(), n: str() }, "stored with another schema at version 1"],
])(
"tx.retire of a cold key stored at %s fails the owner broken as a read would and keeps it (R15, audit)",
async (_, version, schema, message) => {
const store = await freshStore();
// Written by other code: a newer version, or the same version with another schema.
const Other = defineState("notes.note", {
schema: obj(schema),
version,
initial: () => ({ title: "", n: "" }),
keyed: true,
persist: true,
checkpointWhen: () => false,
});
const writer = await storeRestartable(
objects,
(_, self) => {
void self
.state(Other)
.change("a", (draft) => void Object.assign(draft, { title: "kept", n: "1" }), context);
},
{ store },
);
await flushed();
await writer.host.close(context);
const { host, notes, errors, reference } = await persisted(store);
expect(await rejection(notes().retire("a", context))).toMatchObject({
message: expect.stringContaining(message),
});
expect(errors.map(({ kind }) => kind)).toEqual(["object_failed"]);
expect(reference.state.value).toMatchObject({ kind: "failed", cause: { kind: "broken" } });
await host.close(context);
const reader = await storeRestartable(objects, (_, self) => void self.state(Other), { store });
expect(await store.read((executor) => executor.all("SELECT key FROM chord_documents"), context)).toEqual([
{ key: "a" },
]);
await reader.host.close(context);
},
);
it("a retirement cut by its attempt's seal while it finds the document fails no successor (audit)", async () => {
const store = await freshStore();
const Newer = defineState("notes.note", {
schema: obj({ title: str() }),
version: 2,
initial: () => ({ title: "" }),
keyed: true,
persist: true,
checkpointWhen: () => false,
});
const writer = await storeRestartable(
objects,
(_, self) =>
void self.state(Newer).change(
"a",
(draft) => {
draft.title = "kept";
},
context,
),
{ store },
);
await flushed();
await writer.host.close(context);
const { host, notes, errors, reference, fail, retry } = await persisted(store);
// The find's read is held past the seal: the retirement resumes on a cut transaction.
const read = store.read.bind(store);
const entered = Promise.withResolvers<void>();
const gate = Promise.withResolvers<void>();
store.read = async (fn, ctx) => {
store.read = read;
const value = await read(fn, ctx);
entered.resolve();
await gate.promise;
return value;
};
const retired = rejection(notes().retire("a", context));
await entered.promise;
fail();
await retry();
gate.resolve();
expect(await retired).toMatchObject({ code: "released" });
await flushed();
expect(reference.state.value).toEqual({ kind: "live" });
expect(errors.filter(({ kind }) => kind === "object_failed")).toHaveLength(1);
await host.close(context);
});
it("a sealed handle reads cached documents; a cold read is refused released", async () => {
const store = await freshStore();
await seed(store, { a: "x", b: "y" });

5
packages/chord/test/next/documents/persistence.test.ts
View file

@ -167,7 +167,10 @@ describe("loads", () => {
},
undefined,
);
expect(await notes.find("a", context)).toMatchObject({ revision: 1, version: 1 });
expect(await notes.find("a", context)).toMatchObject({
stored: { revision: 1, version: 1 },
fingerprint: expect.any(String),
});
expect(await notes.keys(undefined, 10, context)).toEqual(["a", "b", "\uFFFF", "\u{1F600}"]);
expect(await notes.keys("b", 1, context)).toEqual(["\uFFFF"]);
});

28
packages/chord/test/next/host/ctx.test.ts
View file

@ -5,7 +5,7 @@ import { withLinkedAbortSignal } from "../../../src/next/host/ctx.ts";
import type { ChordHost } from "../../../src/next/host/host.ts";
import type { Ctx, Host, Reference } from "../../../src/next/host/index.ts";
import { constant } from "../../../src/next/state/index.ts";
import { flushed, loaded, rawParent, rejection, type TestPrincipal } from "../support.ts";
import { flushed, incarnationOf, linked, loaded, rawParent, rejection, type TestPrincipal } from "../support.ts";
const Item = local("app.item", { version: 1, params: { id: int() }, members: { id: int() } });
const Desk = local("app.desk", {
@ -140,6 +140,32 @@ describe("ctx", () => {
});
});
describe("local handlers act as the host (§4, audit)", () => {
it("a local object called from a remote handler's ctx.get runs with this host's principal and incarnation", async () => {
const Store = local("app.local-store", { version: 1, params: {}, members: { who: method(undefined, str()) } });
const Gate = remote("app.gate", {
version: 1,
params: {},
variants: { v: { methods: { who: method(undefined, str()), through: method(undefined, str()) } } },
});
const who = (ctx: Ctx<TestPrincipal>) => `${ctx.principal?.kind ?? "none"} ${ctx.incarnation}`;
const { server, client } = await linked((h) => {
h.serve(Store, () => ({ who: async (ctx) => who(ctx) }));
h.serve(Gate, () => ({
variant: "v",
members: {
who: async (ctx) => who(ctx),
through: async (ctx) => (await ctx.get(Store, {})).who(ctx),
},
}));
});
const gate = await client.get(Gate, {}, context);
// The remote handler acts as its caller; the local object it reaches acts as this host, both halves.
expect(await gate.who(context)).toBe(`owner ${incarnationOf(client)}`);
expect(await gate.through(context)).toBe(`server ${incarnationOf(server)}`);
});
});
describe("linked abort signals", () => {
it("abort with the caller's signal or the call's, the caller's reason first", () => {
for (const first of ["caller", "call"] as const) {

19
packages/chord/test/next/host/host.test.ts
View file

@ -694,6 +694,16 @@ describe("acquiring: hold and get", () => {
expect(await refused).toEqual(error);
});
it("invalid params before load resolved: hold throws and get rejects invalid/params at once (audit, 7a)", async () => {
const host = createHost(noErrors);
expect(() => host.hold(Page, { id: "x" } as never, context)).toThrow("params");
expect(await rejection(host.get(Page, { id: "x" } as never, context))).toMatchObject({
code: "invalid",
reason: "params",
});
await host.load([], context);
});
it("a hold released before load resolved makes nothing once it did", async () => {
const factory = vi.fn(({ id }: { id: number }) => ({ id: constant(id) }));
const host = createHost(noErrors);
@ -805,10 +815,10 @@ forEachCodec(() => {
}
/** Sends a hold of `PlacedNotes` (or `type`) for `session`, naming `address` when given. */
function holdOf(peer: RawPeer, session: string, address?: string, type: string = PlacedNotes.name) {
function holdOf(peer: RawPeer, session: string, address?: string, type: string = PlacedNotes.name, hold = 0) {
peer.send({
kind: "hold",
hold: 0,
hold,
type,
version: PlacedNotes.version,
fingerprint: PlacedNotes.fingerprint,
@ -836,9 +846,10 @@ forEachCodec(() => {
hold: 0,
cause: { kind: "broken", error: { kind: "chord", error: misplaced } },
});
holdOf(child, "S2");
// A refused hold is kept (a retry starts it again): the next hold takes the next id.
holdOf(child, "S2", undefined, PlacedNotes.name, 1);
expect(await child.next("failed")).toMatchObject({
hold: 0,
hold: 1,
cause: { kind: "broken", error: { kind: "chord", error: misplaced } },
});
});

59
packages/chord/test/next/host/object.test.ts
View file

@ -311,6 +311,30 @@ describe("dependency recovery (R18)", () => {
]);
});
it("remembers a dependency going live while the dependent activates: the dependent is retried at once (audit)", async () => {
const viewer = view(Core);
let cores = 0;
const { host } = await loaded((h) => {
// Its first factory throws (broken); the second goes live at once, without natives.
h.serve(Core, () => {
if (++cores === 1) throw new Error("first factory");
return { n: constant(2) };
});
h.serve(View, viewer.factory);
});
await flushed();
// The view activates on the core's failure while the core is retried and goes live at once.
const dependency = host.hold(Core, {}, context);
const dependent = host.hold(View, { id: 1 }, context);
dependency.retry();
await flushed();
expect(dependency.state.value).toEqual({ kind: "live" });
expect(dependent.state.value).toEqual({ kind: "live" });
expect(viewer.times).toEqual([0, 0]);
dependent.release();
dependency.release();
});
it("ignores a trigger while an attempt runs: simultaneous triggers give one attempt", async () => {
const gate = Promise.withResolvers<void>();
const svc = flaky((attempt) => attempt === 1, gate.promise);
@ -889,6 +913,41 @@ function nearHold(hold: number): Extract<Message, { kind: "hold" }> {
return { kind: "hold", hold, type, version, fingerprint, params: jsonPart({ session: "S1" }) };
}
describe("a failed activation's cleanup (audit)", () => {
it("closes the natives that opened before it releases its references: a close may use them", async () => {
const Dependent = local("app.dependent", { version: 1, params: { id: int() }, members: { n: int() } });
const seen: unknown[] = [];
const { host, errors } = await loaded((h) => {
h.serve(Core, () => ({ n: constant(5) }));
h.serve(Dependent, (_, self) => {
const core = self.get(Core, {});
self.open(
async () => null,
() => {
try {
seen.push(core.n.value);
} catch (error) {
seen.push(String(error));
}
},
);
self.open(
async () => {
throw broken(new Error("open failed"));
},
() => {},
);
return { n: core.n };
});
});
await expect(host.get(Dependent, { id: 1 }, context)).rejects.toThrow("open failed");
await flushed();
expect(errors.find(({ kind }) => kind === "activation_failed")).toMatchObject({ failure: { kind: "broken" } });
expect(seen).toEqual([5]);
await host.close(context);
});
});
describe("a live object failing (slice 6a §1.4)", () => {
it("fails broken: calls reject with the cause, one object_failed report, natives close, then references go", async () => {
const events: string[] = [];

34
packages/chord/test/next/host/route.test.ts
View file

@ -56,6 +56,40 @@ forEachCodec(() => {
expect((await owner.tab.get(PlacedNotes, { session: "S2" }, context)).variant).toBe("editor");
});
it("ref.retry() holds a refused hold again: authorize admitting it later, over a link and on this host (audit)", async () => {
let admit = false;
const { tab, server } = await relayed({ authorize: () => admit });
const overLink = tab.hold(PlacedNotes, { session: "S1" }, context);
const here = server.hold(PlacedNotes, { session: "S2" }, context);
await transported();
for (const ref of [overLink, here]) {
expect(ref.state.value).toMatchObject({
kind: "failed",
cause: { kind: "broken", error: { code: "forbidden" } },
});
}
admit = true;
overLink.retry();
here.retry();
await transported();
expect(overLink.state.value).toEqual({ kind: "live" });
expect(here.state.value).toEqual({ kind: "live" });
});
it("ref.retry() of a reference refused on this host after host.close ends it released, nothing resolved (audit)", async () => {
let admit = false;
const { server, resolves } = await relayed({ authorize: () => admit });
const ref = server.hold(PlacedNotes, { session: "S1" }, context);
await transported();
expect(ref.state.value).toMatchObject({ kind: "failed", cause: { error: { code: "forbidden" } } });
await server.close(context);
admit = true;
ref.retry();
await transported();
expect(ref.state.value).toMatchObject({ kind: "ended", error: { code: "released", reason: "closed" } });
expect(resolves).toEqual([]);
});
it("fails a hold broken with what authorize throws", async () => {
const { tab } = await relayed({
authorize: () => {

36
packages/chord/test/next/remote/relay.test.ts
View file

@ -15,11 +15,13 @@ import {
fakeTimers,
forEachCodec,
instancesOf,
Notes,
PlacedNotes,
RawPeer,
rawWorkers,
rejection,
relayed,
serveNotes,
type TestPrincipal,
testCodec,
transported,
@ -375,6 +377,40 @@ forEachCodec(() => {
});
});
describe("a parent link that is down (audit)", () => {
it("holds a hold arriving meanwhile on the parent path: it waits loading and binds on the next parent link", async () => {
const server = createHost<TestPrincipal>({ principal: { kind: "server" }, onError: () => {} });
await server.load(
[{ name: "@test/app", facet: defineFacet<TestPrincipal>((h) => void serveNotes(h)) }],
context,
);
const middle = createHost<TestPrincipal>({ principal: { kind: "server" }, onError: () => {} });
await middle.load([], context);
/** Connects the middle host to the server over a new link; returns the server's end. */
const up = (): Link => {
const [a, b] = memoryLinkPair();
server.accept(a, { principal: { kind: "owner" }, codec: testCodec() }, context);
middle.connect(b, { principal: { kind: "server" }, codec: testCodec() }, context);
return a;
};
const first = up();
await transported();
first.close();
await transported();
const tab = createHost<TestPrincipal>({ onError: () => {} });
await tab.load([], context);
const [a, b] = memoryLinkPair();
middle.accept(a, { principal: { kind: "owner" }, codec: testCodec() }, context);
tab.connect(b, { principal: { kind: "server" }, codec: testCodec() }, context);
const notes = tab.hold(Notes, { session: "S1" }, context);
await transported();
expect(notes.state.value).toEqual({ kind: "loading" });
up();
await transported();
expect(notes.state.value).toEqual({ kind: "live" });
});
});
describe("a downstream release", () => {
it("before the upstream hold was sent drops it; after, releases it upstream; messages crossing it are dropped (race 14)", async () => {
let finish: () => void = () => {};

114
packages/chord/test/next/remote/remote-object.test.ts
View file

@ -335,7 +335,7 @@ forEachCodec(() => {
});
describe("self.get through a relay whose upstream dropped (R18)", () => {
it("a loading notice while loading fails the dependent at once, transient; bound retries it", async () => {
it("a loading notice while loading changes nothing: the dependent goes on waiting and goes live when bound", async () => {
const View = local("app.view", { version: 1, params: {}, members: { count: int() } });
const errors: HostError[] = [];
const client = createHost({ onError: (error) => errors.push(error) });
@ -347,19 +347,12 @@ forEachCodec(() => {
);
const loading = client.load([{ name: "@test/view", facet }], context);
const { hold } = await peer.next("hold");
// The relay's upstream link dropped before the hold was bound: no wait for the readyTimeout.
// The relay's upstream link dropped before the hold was bound: a drop is not a dependency failure (audit).
peer.send({ kind: "loading", hold });
await transported();
bindHold(peer, hold, [{ text: "a" }]);
await loading;
expect(errors).toMatchObject([
{
kind: "activation_failed",
object: "app.view",
failure: { kind: "transient", error: { message: expect.stringContaining("app.notes") } },
},
]);
bindHold(peer, hold);
const retried = await bindNext(peer, [{ text: "a" }]);
expect(retried).toBe(1);
expect(errors).toEqual([]);
const view = await client.get(View, {}, context);
expect(view.count.value).toBe(1);
});
@ -631,6 +624,64 @@ forEachCodec(() => {
expect(notes.state.value).toEqual({ kind: "live" });
});
it("ref.retry() of a broken failure sends retry on the same hold: no release, no new hold (audit)", async () => {
const { client, peer } = await rawProvider();
const notes = client.hold(Notes, { session: "S1" }, context);
const { hold } = await peer.next("hold");
const error = { code: "forbidden", message: "not authorized" } as const;
peer.send({ kind: "failed", hold, cause: { kind: "broken", error: { kind: "chord", error } } });
await transported();
expect(notes.state.value).toMatchObject({ kind: "failed", cause: { kind: "broken" } });
notes.retry();
expect(await peer.next("retry")).toEqual({ kind: "retry", hold });
bindHold(peer, hold);
await transported();
expect(notes.state.value).toEqual({ kind: "live" });
expect(peer.received.filter(({ kind }) => kind === "hold" || kind === "release")).toEqual([]);
});
it("ref.retry() of an object whose factory threw, held twice: failed, loading, live; a call made then succeeds (audit)", async () => {
const Flaky = remote("app.flaky", {
version: 1,
params: {},
variants: { only: { state: {}, methods: { ping: method() } } },
});
let factories = 0;
const { client } = await linked((h) =>
h.serve(Flaky, () => {
if (++factories === 1) throw new Error("bad factory");
return { variant: "only", members: { ping: async () => null } };
}),
);
const one = client.hold(Flaky, {}, context);
const two = client.hold(Flaky, {}, context);
await transported();
const states: string[] = [];
const watch = await one.state.watch(context);
states.push(watch.value.kind);
watch.start(async (call) => {
if (call.kind === "value") states.push(call.value.kind);
});
one.retry();
// A call made while it loads again waits for the binding; none sees the old failure.
let called: Promise<string> | undefined;
const unwatch = await one.state.watch(context);
unwatch.start(async (call) => {
if (call.kind === "value" && call.value.kind === "loading" && called === undefined) {
called = one.ping(context).then(
() => "ok",
(error: Error) => `rejected: ${error.message}`,
);
}
});
await transported();
await transported();
expect(states).toEqual(["failed", "loading", "live"]);
expect(await called).toBe("ok");
expect(two.state.value).toEqual({ kind: "live" });
expect(factories).toBe(2);
});
it("a drop while failed: loading, held again on the next link (race 19)", async () => {
const { client, peer } = await rawProvider();
const notes = client.hold(Notes, { session: "S1" }, context);
@ -692,7 +743,7 @@ forEachCodec(() => {
});
});
describe("an activating dependent whose remote dependency's link drops (race 18)", () => {
describe("an activating dependent whose remote dependency's link drops (audit; supersedes race 18)", () => {
fakeTimers();
const View = local("app.view", { version: 1, params: {}, members: { count: int() } });
@ -709,27 +760,36 @@ forEachCodec(() => {
expect((await client.get(View, {}, context, { readyTimeout: 1 })).count.value).toBe(1);
}
it("fails transiently at once with the drop, keeps the reference, and is retried at once when it is bound", async () => {
it("stays loading through the drop, no failure reported, and goes live when the next link binds it", async () => {
const errors: HostError[] = [];
const client = createHost<TestPrincipal>({ onError: (error) => errors.push(error) });
const peer = await rawParent(client);
const loading = client.load([{ name: "@test/view", facet }], context);
await peer.next("hold");
peer.close();
await transported();
const view = client.hold(View, {}, context);
expect(view.state.value).toEqual({ kind: "loading" });
const next = await rawParent(client);
expect(await bindNext(next, [{ text: "a" }])).toBe(0);
await loading;
expect(errors).toMatchObject([
{
kind: "activation_failed",
failure: {
kind: "transient",
error: {
message: "app.view: dependency app.notes(S1) failed",
cause: { message: "the link closed" },
},
},
},
]);
await retried(client, await rawParent(client));
await transported();
expect(view.state.value).toEqual({ kind: "live" });
expect(errors).toEqual([]);
});
it("dropped and never bound again, fails transient at its readyTimeout (the bound of the wait)", async () => {
const errors: HostError[] = [];
const client = createHost<TestPrincipal>({ readyTimeout: 1_000, onError: (error) => errors.push(error) });
const peer = await rawParent(client);
const loading = client.load([{ name: "@test/view", facet }], context);
await peer.next("hold");
peer.close();
await transported();
expect(errors).toEqual([]);
await vi.advanceTimersByTimeAsync(1_000);
await loading;
expect(failures(errors)).toEqual(["app.view: transient"]);
});
it("without any link fails at its readyTimeout, keeps the reference, and is retried when the first link binds it", async () => {