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): make post-cutover reload failures terminal

Browse source
This commit is contained in:
Armin Ronacher 2026-09-01 16:16:12 +02:00
parent c4b0e35abe
commit 5dd8c01320
10 changed files with 323 additions and 247 deletions
Download patch file Download diff file Expand all files Collapse all files

13
packages/chord/PLANNING.md
View file

@ -208,8 +208,7 @@ load replacement module generation
→ run replacement setup
→ validate replacement shape and remote implementations
→ activate replacements in dependency order while the old providers remain routed
→ atomically replace local targets and publish complete remote singleton replacement snapshots
→ drain calls admitted by the retired singleton and keyed instances
→ replace each local or remote singleton directly without withdrawing it
→ deactivate replaced plugins in reverse dependency order
→ dispose the retired loaded generation
```
@ -219,13 +218,13 @@ Properties:
- existing local and remote singleton facades keep object identity;
- captured local and remote methods dispatch to the old provider before cutover and the replacement after cutover;
- replicated-state facades install the replacement snapshot without becoming unhydrated;
- calls admitted before cutover finish against the retired provider while later calls use the replacement;
- every singleton switches directly from its old target to its replacement, but a multi-service reload is not graph-transactional;
- keyed instances from the old plugin close; instances staged by the replacement use fresh generations;
- setup, shape-validation, or replacement-activation failure leaves the active generation unchanged;
- named host resources acquired during activation support overlapping staged replacements, so candidate cleanup restores the old registration and retired cleanup cannot remove the replacement; and
- cutover performs no asynchronous work or facet callbacks.
- any failure after replacement begins terminates the host rather than attempting to preserve a partially transitioned graph.
There is no rollback after cutover. Retired-generation cleanup failures are reported without disconnecting the replacement. Committed application effects are outside the reload transaction.
There is no rollback after cutover. Terminal cleanup revokes every facet handle before best-effort disposal; only pre-cutover candidate cleanup and ordinary host disposal guarantee dependency-ordered cleanup. Committed application effects are outside the reload transaction.
#### Structural replacement
@ -253,9 +252,7 @@ This full-generation path supplies actual plugin load and unload semantics witho
### 6.2 Calls during unload
Replacement atomically stops new admission to the retired service target. Calls already admitted to that target retain it until they settle, while new calls through the same stable facade use the replacement. Old state publication is detached at cutover and cannot update the replacement replica.
Facet deactivation and loaded-module disposal wait for admitted calls to drain. A future killable-isolate host may force termination during abnormal shutdown, but ordinary reload does not reject admitted calls. Transport disconnect still cancels only connection-owned invocations; it does not imply application-level cancellation or rollback.
Replacement never leaves a singleton facade without a target: an invocation resolves either the old implementation or its replacement. Work already running in a retired facet is not drained; later use of that facet's revoked handles may fail as stale work. A future killable-isolate host terminates such work directly. Transport disconnect still cancels only connection-owned invocations; it does not imply application-level cancellation or rollback.
### 6.3 Update failure reporting

6
packages/chord/README.md
View file

@ -194,9 +194,9 @@ receiving host.
To reload, load a candidate, pass its facets to `FacetHost.reload()`, dispose the
candidate on failure, and dispose the retired `LoadedFacets` only after a
successful cutover. The host activates and validates the candidate while the old
providers remain routed, atomically replaces singleton targets, then drains calls
admitted by the retired targets before resolving reload. Stable service handles
therefore do not become unavailable during an ordinary reload. Keyed instances
providers remain routed, then replaces each singleton directly without an
unavailable interval. Stable service handles therefore do not become disconnected
during an ordinary reload. Keyed instances
remain incarnation-specific and replacements receive fresh generations. The
bundler writes a complete temporary directory before replacing the previous
output, so loaders do not observe partially built generations.

127
packages/chord/src/facets/host.ts
View file

@ -90,6 +90,10 @@ class FacetLifecycle {
}
}
revoke(): void {
this.#serviceAccess = false;
}
own(disposal: Disposal): void {
this.assertRunning("own resources");
this.#effects.push(disposal);
@ -174,23 +178,18 @@ class LocalKeyedServiceRegistry implements KeyedServiceSource {
}
const generation = (registration.generations.get(key) ?? 0) + 1;
registration.generations.set(key, generation);
const slot = new ServiceSlot(service.id, false);
void slot.bind(implementation as object);
const instance: InstanceDirectoryEntry = {
key,
generation,
service: slot.view<T>(() => {}) as object,
service: implementation as object,
deactivate() {},
};
registration.directory.insert(instance);
let closed = false;
let retired: Promise<void> | undefined;
return () => {
if (closed) return retired;
if (closed) return;
closed = true;
registration.directory.remove(instance);
retired = slot.unbind();
return retired;
};
}
@ -244,7 +243,7 @@ class HostServiceSlots {
let stopped = false;
const stop = source.observe(service, (target, context) => {
const slot = new ServiceSlot(service.id, !service.local);
void slot.bind(target as object);
slot.bind(target as object);
return handler(
slot.view<T>(() => {
assertAccess();
@ -262,8 +261,8 @@ class HostServiceSlots {
};
}
bindSingleton(serviceId: string, target: object): Promise<void> {
return this.#singletons.get(serviceId)?.bind(target) ?? Promise.resolve();
bindSingleton(serviceId: string, target: object): void {
this.#singletons.get(serviceId)?.bind(target);
}
bindKeyed(serviceId: string, services: KeyedServiceSource): void {
@ -271,19 +270,18 @@ class HostServiceSlots {
}
dispose(): void {
for (const slot of this.#singletons.values()) void slot.unbind();
for (const slot of this.#singletons.values()) slot.unbind();
this.#singletons.clear();
this.#keyedSources.clear();
}
}
type ServiceInstanceInstaller<T> = (key: string, implementation: T) => Disposal;
type ServiceInstanceInstaller<T> = (key: string, implementation: T) => () => void;
interface StagedServiceInstance<T> {
readonly key: string;
readonly implementation: T;
release?: Disposal;
retirement?: Promise<void>;
release?: () => void;
}
class StagedServiceSpawner<T> implements ServiceSpawner<T> {
@ -312,12 +310,10 @@ class StagedServiceSpawner<T> implements ServiceSpawner<T> {
const instance: StagedServiceInstance<T> = { key, implementation };
this.#instances.set(key, instance);
if (this.#installer !== undefined) instance.release = this.#installer(key, implementation);
const close = (): void | Promise<void> => {
if (this.#instances.get(key) !== instance) return instance.retirement;
const close = (): void => {
if (this.#instances.get(key) !== instance) return;
this.#instances.delete(key);
const result = instance.release?.();
if (isPromiseLike(result)) instance.retirement = Promise.resolve(result);
return instance.retirement;
instance.release?.();
};
this.#lifecycle.own(close);
return close;
@ -331,7 +327,7 @@ type FacetProvision =
readonly implementation: object;
install(provider: RemoteServiceProvider): void;
validateReplacement(provider: RemoteServiceProvider): void;
replace(provider: RemoteServiceProvider): Promise<void>;
replace(provider: RemoteServiceProvider): void;
}
| {
readonly kind: "keyed";
@ -450,58 +446,68 @@ export class FacetKernel {
}
} catch (error) {
const cleanupErrors = await disposeFacetRecords(staged.reverse());
this.#phase = "active";
if (cleanupErrors.length > 0) {
throw new AggregateError([error, ...cleanupErrors], "Facet reload setup and cleanup failed");
const abortErrors = await this.#abort();
throw new AggregateError(
[error, ...cleanupErrors, ...abortErrors],
"Facet reload setup and cleanup failed",
);
}
this.#phase = "active";
throw error;
}
const replacements = new Map(candidates.map((record) => [record.facetId, record]));
const candidateOrder = this.#activationOrder.flatMap((id) => {
const candidate = replacements.get(id);
return candidate === undefined ? [] : [candidate];
});
try {
for (const id of this.#activationOrder) {
await replacements.get(id)?.lifecycle.activate();
}
for (const candidate of candidateOrder) await candidate.lifecycle.activate();
for (const candidate of candidateOrder) this.#validateReplacementProvisions(candidate.provisions);
} catch (error) {
const cleanupErrors = await disposeFacetRecords([...candidates].reverse());
this.#phase = "active";
const cleanupErrors = await disposeFacetRecords([...candidateOrder].reverse());
if (cleanupErrors.length > 0) {
throw new AggregateError([error, ...cleanupErrors], "Facet reload activation and cleanup failed");
const abortErrors = await this.#abort();
throw new AggregateError(
[error, ...cleanupErrors, ...abortErrors],
"Facet reload activation and cleanup failed",
);
}
this.#phase = "active";
throw error;
}
const previous = new Map(candidates.map(({ facetId }) => [facetId, this.#facets.get(facetId)!] as const));
const retirements = candidates.flatMap(({ provisions }) =>
provisions.flatMap((provision) => {
if (provision.kind !== "singleton") return [];
return [
provision.service.local
? this.#serviceSlots.bindSingleton(provision.service.id, provision.implementation)
: provision.replace(this.provider),
];
}),
);
for (const candidate of candidates) this.#facets.set(candidate.facetId, candidate);
await Promise.all(retirements);
const retirementErrors = await disposeFacetRecords(
[...this.#activationOrder].reverse().flatMap((id) => {
const record = previous.get(id);
return record === undefined ? [] : [record];
}),
);
for (const candidate of candidates) {
for (const provision of candidate.provisions) {
if (provision.kind !== "keyed") continue;
if (provision.service.local) provision.connectLocal(this.#localKeyedRegistry);
else provision.connectRemote(this.provider);
const previous = candidateOrder.map(({ facetId }) => this.#facets.get(facetId)!);
for (const candidate of candidateOrder) this.#facets.set(candidate.facetId, candidate);
try {
for (const candidate of candidateOrder) {
for (const provision of candidate.provisions) {
if (provision.kind !== "singleton") continue;
if (provision.service.local) {
this.#serviceSlots.bindSingleton(provision.service.id, provision.implementation);
} else {
provision.replace(this.provider);
}
}
}
const retirementErrors = await disposeFacetRecords(previous.reverse());
if (retirementErrors.length === 1) throw retirementErrors[0];
if (retirementErrors.length > 1) {
throw new AggregateError(retirementErrors, "Failed to retire replaced facets");
}
for (const candidate of candidateOrder) {
for (const provision of candidate.provisions) {
if (provision.kind !== "keyed") continue;
if (provision.service.local) provision.connectLocal(this.#localKeyedRegistry);
else provision.connectRemote(this.provider);
}
}
} catch (error) {
const abortErrors = await this.#abort(previous);
throw new AggregateError([error, ...abortErrors], "Facet reload failed after cutover");
}
this.#phase = "active";
if (retirementErrors.length === 1) throw retirementErrors[0];
if (retirementErrors.length > 1) {
throw new AggregateError(retirementErrors, "Failed to retire replaced facets");
}
}
async dispose(): Promise<void> {
@ -738,9 +744,16 @@ export class FacetKernel {
}
}
async #terminate(): Promise<unknown[]> {
async #abort(extraRecords: readonly FacetRuntime[] = []): Promise<unknown[]> {
for (const record of this.#facets.values()) record.lifecycle.revoke();
for (const record of extraRecords) record.lifecycle.revoke();
return this.#terminate(extraRecords);
}
async #terminate(extraRecords: readonly FacetRuntime[] = []): Promise<unknown[]> {
this.#phase = "disposing";
const errors = await this.#disposeLifecycles();
errors.push(...(await disposeFacetRecords([...extraRecords].reverse())));
try {
this.#localKeyedServices?.dispose();
} catch (error) {

60
packages/chord/src/services/calls.ts
View file

@ -1,60 +0,0 @@
type Resolve = () => void;
/** Tracks calls admitted to one service target so replacement can drain it without disconnecting consumers. */
export class CallTracker {
#active = 0;
#retired = false;
#settled: Promise<void> | undefined;
#resolve: Resolve | undefined;
run<T>(callback: () => T): T {
if (this.#retired) throw new Error("Service target is retired");
this.#active += 1;
let result: T;
try {
result = callback();
} catch (error) {
this.#release();
throw error;
}
if (!isPromiseLike(result)) {
this.#release();
return result;
}
return Promise.resolve(result).then(
(value) => {
this.#release();
return value;
},
(error: unknown) => {
this.#release();
throw error;
},
) as T;
}
retire(): Promise<void> {
if (this.#retired) return this.#settled ?? Promise.resolve();
this.#retired = true;
if (this.#active === 0) return Promise.resolve();
this.#settled = new Promise<void>((resolve) => {
this.#resolve = resolve;
});
return this.#settled;
}
#release(): void {
this.#active -= 1;
if (!this.#retired || this.#active !== 0) return;
this.#resolve?.();
this.#resolve = undefined;
}
}
function isPromiseLike(value: unknown): value is PromiseLike<unknown> {
return (
((typeof value === "object" && value !== null) || typeof value === "function") &&
"then" in value &&
typeof value.then === "function"
);
}

36
packages/chord/src/services/handle.ts
View file

@ -1,14 +1,6 @@
import { CallTracker } from "./calls.ts";
interface ResolvedValue {
readonly value: unknown;
readonly receiver: object;
readonly calls: CallTracker;
}
interface ServiceTarget {
readonly implementation: object;
readonly calls: CallTracker;
}
type ValueResolver = () => ResolvedValue;
@ -17,7 +9,7 @@ type ValueResolver = () => ResolvedValue;
export class ServiceSlot {
readonly #serviceId: string;
readonly #wrapObjects: boolean;
#target: ServiceTarget | undefined;
#implementation: object | undefined;
constructor(serviceId: string, wrapObjects: boolean) {
this.#serviceId = serviceId;
@ -28,26 +20,21 @@ export class ServiceSlot {
return new ServiceView(this, assertAccess, this.#wrapObjects).proxy as T;
}
bind(implementation: object): Promise<void> {
const previous = this.#target;
this.#target = { implementation, calls: new CallTracker() };
return previous?.calls.retire() ?? Promise.resolve();
bind(implementation: object): void {
this.#implementation = implementation;
}
unbind(): Promise<void> {
const previous = this.#target;
this.#target = undefined;
return previous?.calls.retire() ?? Promise.resolve();
unbind(): void {
this.#implementation = undefined;
}
resolve(property: PropertyKey, assertAccess: () => void): ResolvedValue {
assertAccess();
const target = this.#target;
if (target === undefined) throw new Error(`Service ${this.#serviceId} is disconnected`);
const implementation = this.#implementation;
if (implementation === undefined) throw new Error(`Service ${this.#serviceId} is disconnected`);
return {
value: Reflect.get(target.implementation, property, target.implementation),
receiver: target.implementation,
calls: target.calls,
value: Reflect.get(implementation, property, implementation),
receiver: implementation,
};
}
}
@ -96,9 +83,9 @@ class ValueView {
}
#invoke(args: unknown[]): unknown {
const { value, receiver, calls } = this.#resolve();
const { value, receiver } = this.#resolve();
if (typeof value !== "function") throw new TypeError("Service member is not callable");
return calls.run(() => Reflect.apply(value, receiver, args));
return Reflect.apply(value, receiver, args);
}
#get(property: PropertyKey): unknown {
@ -108,7 +95,6 @@ class ValueView {
return {
value: Reflect.get(parent.value, property, parent.value),
receiver: parent.value,
calls: parent.calls,
};
};
const current = resolve().value;

56
packages/chord/src/services/provider.ts
View file

@ -13,7 +13,6 @@ import type {
ServiceSubscription,
ServiceSubscriptionSnapshot,
} from "../types.ts";
import { CallTracker } from "./calls.ts";
import { RemoteServiceError } from "./errors.ts";
import { serviceDeliveryContext } from "./state.ts";
import { getReplicatedStateInternals, type ReplicatedStateInternals } from "./state-internals.ts";
@ -37,7 +36,6 @@ interface ProviderInstance {
readonly implementation: object;
readonly members: ReadonlyMap<string, InstanceMember>;
readonly removeMemberListeners: readonly (() => void)[];
readonly calls: CallTracker;
active: boolean;
}
@ -125,20 +123,18 @@ export class RemoteServiceProvider {
registration.singletonShape = shape;
}
/** Disconnect one singleton and resolve after calls admitted by the retired target settle. */
withdraw<T>(service: Service<T>): Promise<void> {
/** Disconnect one singleton while preserving active subscriptions and remote facades. */
withdraw<T>(service: Service<T>): void {
this.#assertActive();
this.#assertRemotable(service);
this.#assertAllowed(service.id);
const registration = this.#registration(service.id, "singleton");
const previous = registration.singleton;
if (previous === undefined) return Promise.resolve();
if (previous === undefined) return;
previous.active = false;
for (const remove of previous.removeMemberListeners) remove();
delete registration.singleton;
const retired = previous.calls.retire();
this.#emit(registration, { type: "unavailable" });
return retired;
}
/** Check a singleton replacement without changing the active provider. */
@ -151,8 +147,8 @@ export class RemoteServiceProvider {
this.#assertSingletonShape(registration, serviceMemberShape(classified.members));
}
/** Replace one singleton immediately and resolve after calls admitted by the retired target settle. */
replace<T>(service: Service<T>, implementation: NoInfer<RemoteServiceContract<T>>): Promise<void> {
/** Replace one singleton without making its stable remote facade unavailable. */
replace<T>(service: Service<T>, implementation: NoInfer<RemoteServiceContract<T>>): void {
this.#assertActive();
this.#assertRemotable(service);
this.#assertAllowed(service.id);
@ -168,9 +164,7 @@ export class RemoteServiceProvider {
}
registration.singleton = replacement;
registration.singletonShape = shape;
const retired = previous?.calls.retire() ?? Promise.resolve();
this.#emit(registration, { type: "replaced", snapshot: this.#snapshotInstance(replacement) });
return retired;
}
use<T>(service: Service<T>): T {
@ -204,17 +198,14 @@ export class RemoteServiceProvider {
registration.instances.set(key, instance);
this.#emit(registration, { type: "spawned", instance: this.#snapshotInstance(instance) });
let closed = false;
let retired: Promise<void> | undefined;
return () => {
if (closed) return retired;
if (closed) return;
closed = true;
if (registration.instances.get(key) !== instance) return undefined;
if (registration.instances.get(key) !== instance) return;
instance.active = false;
for (const remove of instance.removeMemberListeners) remove();
registration.instances.delete(key);
retired = instance.calls.retire();
this.#emit(registration, { type: "closed", instance: address });
return retired;
};
}
@ -239,9 +230,7 @@ export class RemoteServiceProvider {
`Remote service member ${call.serviceId}.${call.member} is not a method`,
);
}
const result: unknown = await instance.calls.run(() =>
Reflect.apply(member.method, instance.implementation, [...call.args, context]),
);
const result: unknown = await Reflect.apply(member.method, instance.implementation, [...call.args, context]);
return result as JsonValue | undefined;
}
@ -271,11 +260,19 @@ export class RemoteServiceProvider {
activate: () => {
if (subscriber.closed || subscriber.active) return;
subscriber.active = true;
const errors: unknown[] = [];
try {
for (const entry of subscriber.buffer.splice(0)) listener(entry.update, entry.context);
for (const entry of subscriber.buffer.splice(0)) {
try {
listener(entry.update, entry.context);
} catch (error) {
errors.push(error);
}
}
} finally {
if (subscriber.terminated) subscriber.closed = true;
}
throwCollectedErrors(errors, "Failed to activate remote service subscription");
},
close: () => {
if (subscriber.closed) return;
@ -352,7 +349,6 @@ export class RemoteServiceProvider {
implementation: classified.implementation,
members: classified.members,
removeMemberListeners,
calls: new CallTracker(),
active: true,
};
for (const [name, member] of classified.members) {
@ -471,12 +467,21 @@ export class RemoteServiceProvider {
#emit(registration: ServiceRegistration, update: ServiceProviderUpdate, context?: Context): void {
if (registration.subscribers.size === 0) return;
const deliveryContext = context ?? serviceDeliveryContext();
const errors: unknown[] = [];
for (const subscriber of registration.subscribers) {
if (subscriber.closed) continue;
const entry = { update, context: deliveryContext };
if (subscriber.active) subscriber.listener(entry.update, entry.context);
else subscriber.buffer.push(entry);
if (!subscriber.active) {
subscriber.buffer.push(entry);
continue;
}
try {
subscriber.listener(entry.update, entry.context);
} catch (error) {
errors.push(error);
}
}
throwCollectedErrors(errors, `Failed to publish remote service ${registration.serviceId} update`);
}
#assertRemotable(service: { readonly id: string; readonly local: boolean }): void {
@ -564,6 +569,11 @@ function classifyRemoteServiceImplementation(
return { implementation, members };
}
function throwCollectedErrors(errors: readonly unknown[], message: string): void {
if (errors.length === 1) throw errors[0];
if (errors.length > 1) throw new AggregateError(errors, message);
}
function serviceMemberShape(members: ReadonlyMap<string, InstanceMember>): ReadonlyMap<string, ServiceMemberKind> {
return new Map([...members].map(([name, member]) => [name, member.kind]));
}

2
packages/chord/src/types.ts
View file

@ -246,7 +246,7 @@ export interface FacetOptions {
export interface FacetHost {
readonly services: RemoteServiceProvider;
/** Activate and atomically replace facets with matching IDs without disconnecting consumer service handles. */
/** Activate and replace facets with matching IDs without disconnecting consumer service handles. */
reload(facets: readonly Facet[]): Promise<void>;
dispose(): Promise<void>;
}

164
packages/chord/test/facet-loader.test.ts
View file

@ -202,76 +202,6 @@ describe("facet loader", () => {
]);
});
test("drains calls admitted by the old provider without blocking calls to the replacement", async () => {
let localValue: GenerationValue | undefined;
let oldCalls = 0;
let reportOldCallsStarted!: () => void;
const oldCallsStarted = new Promise<void>((resolve) => {
reportOldCallsStarted = resolve;
});
let finishOldCalls!: () => void;
const oldCallsCanFinish = new Promise<void>((resolve) => {
finishOldCalls = resolve;
});
let reportReplacementActivated!: () => void;
const replacementActivated = new Promise<void>((resolve) => {
reportReplacementActivated = resolve;
});
const provider = (name: string, block: boolean) =>
defineFacet({
id: "draining-provider",
setup(env) {
const implementation: GenerationValue = {
async read() {
if (block) {
oldCalls += 1;
if (oldCalls === 2) reportOldCallsStarted();
await oldCallsCanFinish;
}
return name;
},
};
env.provide(LocalGenerationValue, implementation);
env.provide(RemoteGenerationValue, implementation);
if (!block) env.onActivate(reportReplacementActivated);
},
});
const consumer = defineFacet({
id: "draining-consumer",
setup(env) {
localValue = env.use(LocalGenerationValue);
},
});
const host = await createFacetHost({ facets: [consumer, provider("A", true)] });
const remoteServices = createRemoteServiceBinding({
services: [RemoteGenerationValue],
transport: createLoopbackServiceTransport(host.services),
});
const remoteValue = remoteServices.use(RemoteGenerationValue);
await remoteServices.ready(BACKGROUND_CONTEXT);
const oldLocal = localValue!.read(BACKGROUND_CONTEXT);
const oldRemote = remoteValue.read(BACKGROUND_CONTEXT);
await oldCallsStarted;
let reloadSettled = false;
const reload = host.reload([provider("B", false)]).finally(() => {
reloadSettled = true;
});
await replacementActivated;
await new Promise((resolve) => setTimeout(resolve, 0));
await expect(localValue!.read(BACKGROUND_CONTEXT)).resolves.toBe("B");
await expect(remoteValue.read(BACKGROUND_CONTEXT)).resolves.toBe("B");
expect(reloadSettled).toBe(false);
finishOldCalls();
await expect(oldLocal).resolves.toBe("A");
await expect(oldRemote).resolves.toBe("A");
await reload;
await remoteServices.dispose(BACKGROUND_CONTEXT);
await host.dispose();
});
test("rejects remote singleton member shape changes before reload cutover", async () => {
let retained: GenerationValue | undefined;
let providerDisposed = false;
@ -317,6 +247,100 @@ describe("facet loader", () => {
expect(providerDisposed).toBe(true);
});
test("terminates the host when old cleanup fails after cutover", async () => {
const cleanupFailure = new Error("cleanup failed");
const provider = (name: string, failCleanup: boolean) =>
defineFacet({
id: "cleanup-provider",
setup(env) {
env.provide(RemoteGenerationValue, {
async read() {
return name;
},
});
env.onDeactivate(() => {
if (failCleanup) throw cleanupFailure;
});
},
});
const host = await createFacetHost({ facets: [provider("A", true)] });
await expect(host.reload([provider("B", false)])).rejects.toThrow("Facet reload failed after cutover");
await expect(host.reload([])).rejects.toThrow("Facet host cannot reload while dead");
await host.dispose();
});
test("cleans failed candidate activation in reverse dependency order", async () => {
const failure = new Error("consumer activation failed");
const trace: string[] = [];
const provider = (name: string) =>
defineFacet({
id: "ordered-provider",
setup(env) {
env.provide(RemoteGenerationValue, {
async read() {
return name;
},
});
env.onActivate(() => {
trace.push(`activate provider ${name}`);
});
env.onDeactivate(() => {
trace.push(`deactivate provider ${name}`);
});
},
});
const consumer = (name: string, fail: boolean) =>
defineFacet({
id: "ordered-consumer",
setup(env) {
env.use(RemoteGenerationValue);
env.onActivate(() => {
trace.push(`activate consumer ${name}`);
if (fail) throw failure;
});
env.onDeactivate(() => {
trace.push(`deactivate consumer ${name}`);
});
},
});
const host = await createFacetHost({ facets: [consumer("A", false), provider("A")] });
trace.length = 0;
await expect(host.reload([consumer("B", true), provider("B")])).rejects.toBe(failure);
expect(trace).toEqual([
"activate provider B",
"activate consumer B",
"deactivate consumer B",
"deactivate provider B",
]);
await host.dispose();
});
test("terminates the host when replacement publication fails after cutover", async () => {
const publicationFailure = new Error("publication failed");
const provider = (name: string) =>
defineFacet({
id: "publication-provider",
setup(env) {
env.provide(RemoteGenerationValue, {
async read() {
return name;
},
});
},
});
const host = await createFacetHost({ facets: [provider("A")] });
const subscription = host.services.subscribe(RemoteGenerationValue.id, "singleton", () => {
throw publicationFailure;
});
subscription.activate();
await expect(host.reload([provider("B")])).rejects.toThrow("Facet reload failed after cutover");
await expect(host.reload([])).rejects.toThrow("Facet host cannot reload while dead");
await host.dispose();
});
test("keeps the old generation active when replacement activation fails before cutover", async () => {
const failure = new Error("replacement activation failed");
const trace: string[] = [];

55
packages/chord/test/facets.test.ts
View file

@ -196,6 +196,61 @@ describe("facet host", () => {
expect(observed[1]!.context.abortSignal?.aborted).toBe(true);
});
test("terminates the host when keyed replacement publication fails", async () => {
const failure = new Error("spawn publication failed");
const provider = (value: string) =>
defineFacet({
id: "failing-keyed-provider",
setup(env) {
const values = env.provideMany(KeyedValue);
env.onActivate(() => {
values.spawn("current", {
async read() {
return value;
},
});
});
},
});
const host = await createFacetHost({ facets: [provider("A")] });
const subscription = host.services.subscribe(KeyedValue.id, "keyed", (update) => {
if (update.type === "spawned") throw failure;
});
subscription.activate();
await expect(host.reload([provider("B")])).rejects.toThrow("Facet reload failed after cutover");
await expect(host.reload([])).rejects.toThrow("Facet host cannot reload while dead");
expect(() => host.services.use(KeyedValue)).toThrow("Remote service provider is disposed");
await host.dispose();
});
test("terminates the host when keyed retirement publication fails", async () => {
const failure = new Error("close publication failed");
const provider = (value: string) =>
defineFacet({
id: "failing-keyed-retirement-provider",
setup(env) {
const values = env.provideMany(KeyedValue);
env.onActivate(() => {
values.spawn("current", {
async read() {
return value;
},
});
});
},
});
const host = await createFacetHost({ facets: [provider("A")] });
const subscription = host.services.subscribe(KeyedValue.id, "keyed", (update) => {
if (update.type === "closed") throw failure;
});
subscription.activate();
await expect(host.reload([provider("B")])).rejects.toThrow("Facet reload failed after cutover");
await expect(host.reload([])).rejects.toThrow("Facet host cannot reload while dead");
await host.dispose();
});
test("keeps unrestricted local keyed services process-local across provider reloads", async () => {
const observed: Array<{ service: LocalKeyedValue; context: Context }> = [];
const consumer = defineFacet({

51
packages/chord/test/services.test.ts
View file

@ -309,6 +309,57 @@ describe("remote services", () => {
provider.dispose();
});
test("delivers active subscriber updates before reporting listener failures", () => {
const failure = new Error("listener failed");
const provider = new RemoteServiceProvider([Models]);
provider.provide(Models, {
state: replicatedState<ModelsState>({ selected: null, revision: 1 }),
async select() {},
});
let delivered = 0;
const failing = provider.subscribe(Models.id, "singleton", () => {
throw failure;
});
const succeeding = provider.subscribe(Models.id, "singleton", () => {
delivered += 1;
});
failing.activate();
succeeding.activate();
expect(() =>
provider.replace(Models, {
state: replicatedState<ModelsState>({ selected: null, revision: 2 }),
async select() {},
}),
).toThrow(failure);
expect(delivered).toBe(1);
failing.close();
succeeding.close();
provider.dispose();
});
test("replays every buffered update before reporting listener failures", () => {
const failure = new Error("listener failed");
const provider = new RemoteServiceProvider([Models]);
const state = replicatedState<ModelsState>({ selected: null, revision: 0 });
provider.provide(Models, { state, async select() {} });
let delivered = 0;
const subscription = provider.subscribe(Models.id, "singleton", () => {
delivered += 1;
throw failure;
});
state.state.revision = 1;
state.publish(BACKGROUND_CONTEXT);
state.state.revision = 2;
state.publish(BACKGROUND_CONTEXT);
expect(() => subscription.activate()).toThrow("Failed to activate remote service subscription");
expect(delivered).toBe(2);
subscription.close();
provider.dispose();
});
test("clears retained facades when providers and bindings are disposed", async () => {
const provider = new RemoteServiceProvider([Models]);
provider.provide(Models, {