diff --git a/packages/chord/src/next/documents/persistence.ts b/packages/chord/src/next/documents/persistence.ts index 3efc45308..46aebb93e 100644 --- a/packages/chord/src/next/documents/persistence.ts +++ b/packages/chord/src/next/documents/persistence.ts @@ -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 { + /** + * 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 { 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) { diff --git a/packages/chord/src/next/documents/transaction.ts b/packages/chord/src/next/documents/transaction.ts index 506c15468..af47dd2e4 100644 --- a/packages/chord/src/next/documents/transaction.ts +++ b/packages/chord/src/next/documents/transaction.ts @@ -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; + } +} diff --git a/packages/chord/src/next/host/attempt.ts b/packages/chord/src/next/host/attempt.ts index a424b5ff6..b6d04a898 100644 --- a/packages/chord/src/next/host/attempt.ts +++ b/packages/chord/src/next/host/attempt.ts @@ -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(); host.count("calls", 1); diff --git a/packages/chord/src/next/host/host.ts b/packages/chord/src/next/host/host.ts index ecfad05ed..cbafd799b 100644 --- a/packages/chord/src/next/host/host.ts +++ b/packages/chord/src/next/host/host.ts @@ -181,8 +181,9 @@ export interface Host

(Type: C, params: ParamsOf, context: Context, options?: GetOptions): Promise>; /** * 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(Type: C, params: ParamsOf, context: Context, options?: GetOptions): Held; /** @@ -641,9 +642,9 @@ export class ChordHost

+ this.#state === "closed" + ? binding.end(this.#closedError()) + : this.#bindOrFail(binding, Type, checked, context); + binding.refuse({ kind: "broken", error: asChordError(error, Type.name) }, again); } } diff --git a/packages/chord/src/next/host/object.ts b/packages/chord/src/next/host/object.ts index 43e4d80fd..f422a58fb 100644 --- a/packages/chord/src/next/host/object.ts +++ b/packages/chord/src/next/host/object.ts @@ -125,6 +125,8 @@ export class ChordObject implements Target { readonly #holders = new Set(); /** Objects this object's activation waits for; followed to find cycles (Q5). */ readonly #waitingFor = new Set(); + /** 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. */ diff --git a/packages/chord/src/next/host/placement.ts b/packages/chord/src/next/host/placement.ts index 67f4215ca..e6011168f 100644 --- a/packages/chord/src/next/host/placement.ts +++ b/packages/chord/src/next/host/placement.ts @@ -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 }; diff --git a/packages/chord/src/next/host/reference.ts b/packages/chord/src/next/host/reference.ts index 142dae425..4b9ed1ff8 100644 --- a/packages/chord/src/next/host/reference.ts +++ b/packages/chord/src/next/host/reference.ts @@ -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. */ diff --git a/packages/chord/src/next/jobs/persisted.ts b/packages/chord/src/next/jobs/persisted.ts index 9802ad441..6530ec58d 100644 --- a/packages/chord/src/next/jobs/persisted.ts +++ b/packages/chord/src/next/jobs/persisted.ts @@ -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 { - 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). */ diff --git a/packages/chord/src/next/remote/hold.ts b/packages/chord/src/next/remote/hold.ts index 232d2122a..59ee856c9 100644 --- a/packages/chord/src/next/remote/hold.ts +++ b/packages/chord/src/next/remote/hold.ts @@ -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 { 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(); } } diff --git a/packages/chord/src/next/remote/path.ts b/packages/chord/src/next/remote/path.ts index b1050ec7d..77efb0755 100644 --- a/packages/chord/src/next/remote/path.ts +++ b/packages/chord/src/next/remote/path.ts @@ -18,6 +18,8 @@ export class Path { readonly #holders = new Set(); 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); } diff --git a/packages/chord/src/next/remote/remote-object.ts b/packages/chord/src/next/remote/remote-object.ts index 450c051e1..1765b7d1d 100644 --- a/packages/chord/src/next/remote/remote-object.ts +++ b/packages/chord/src/next/remote/remote-object.ts @@ -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). */ diff --git a/packages/chord/test/next/documents/keyed-slot.test.ts b/packages/chord/test/next/documents/keyed-slot.test.ts index 674ff1e85..fbc3c2a3e 100644 --- a/packages/chord/test/next/documents/keyed-slot.test.ts +++ b/packages/chord/test/next/documents/keyed-slot.test.ts @@ -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(); + const gate = Promise.withResolvers(); + 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" }); diff --git a/packages/chord/test/next/documents/persistence.test.ts b/packages/chord/test/next/documents/persistence.test.ts index 13fe83ea8..b4be944bc 100644 --- a/packages/chord/test/next/documents/persistence.test.ts +++ b/packages/chord/test/next/documents/persistence.test.ts @@ -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"]); }); diff --git a/packages/chord/test/next/host/ctx.test.ts b/packages/chord/test/next/host/ctx.test.ts index 0d4cc480b..732ed2d3a 100644 --- a/packages/chord/test/next/host/ctx.test.ts +++ b/packages/chord/test/next/host/ctx.test.ts @@ -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) => `${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) { diff --git a/packages/chord/test/next/host/host.test.ts b/packages/chord/test/next/host/host.test.ts index 07c4eb751..64a291b43 100644 --- a/packages/chord/test/next/host/host.test.ts +++ b/packages/chord/test/next/host/host.test.ts @@ -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 } }, }); }); diff --git a/packages/chord/test/next/host/object.test.ts b/packages/chord/test/next/host/object.test.ts index 2df9253e8..d492240f0 100644 --- a/packages/chord/test/next/host/object.test.ts +++ b/packages/chord/test/next/host/object.test.ts @@ -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(); const svc = flaky((attempt) => attempt === 1, gate.promise); @@ -889,6 +913,41 @@ function nearHold(hold: number): Extract { 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[] = []; diff --git a/packages/chord/test/next/host/route.test.ts b/packages/chord/test/next/host/route.test.ts index dc4485504..eb3d6413f 100644 --- a/packages/chord/test/next/host/route.test.ts +++ b/packages/chord/test/next/host/route.test.ts @@ -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: () => { diff --git a/packages/chord/test/next/remote/relay.test.ts b/packages/chord/test/next/remote/relay.test.ts index 78fbddb0c..701b2adaf 100644 --- a/packages/chord/test/next/remote/relay.test.ts +++ b/packages/chord/test/next/remote/relay.test.ts @@ -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({ principal: { kind: "server" }, onError: () => {} }); + await server.load( + [{ name: "@test/app", facet: defineFacet((h) => void serveNotes(h)) }], + context, + ); + const middle = createHost({ 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({ 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 = () => {}; diff --git a/packages/chord/test/next/remote/remote-object.test.ts b/packages/chord/test/next/remote/remote-object.test.ts index e5e3d69b7..03baaa01c 100644 --- a/packages/chord/test/next/remote/remote-object.test.ts +++ b/packages/chord/test/next/remote/remote-object.test.ts @@ -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 | 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({ 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({ 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 () => {