Add durable TypeScript provider sessions
This commit is contained in:
Meta Vens
parent
b943fa7ef0
commit
18cebf740f
8 changed files with 1905 additions and 2 deletions
|
|
@ -262,6 +262,44 @@ Schema and shard catalog factories delegate to `cultnet-ts`; the branded
|
|||
`CultMesh` entrypoint does not create a second owner for schema discovery or
|
||||
topology truth.
|
||||
|
||||
## Long-Lived Providers
|
||||
|
||||
TypeScript provider daemons use `CultMeshProviderSession` to retain one logical
|
||||
provider lifecycle across physical reconnects. The session owns registration,
|
||||
lease renewal, desired typed publication replay, command dispatch, receipt
|
||||
publication, and withdrawal. Provider code owns domain state and idempotent
|
||||
command transactions; the injected transport owns CultNet wire details.
|
||||
|
||||
```ts
|
||||
import { CultMeshProviderSession } from "cultmesh-ts";
|
||||
|
||||
const session = new CultMeshProviderSession({
|
||||
identity: {
|
||||
providerId: "voidbot.swarm",
|
||||
serviceInstanceId: "voidbot-worker-7",
|
||||
endpointId: "odin:voidbot-worker-7",
|
||||
verseId: "voidbot.local",
|
||||
},
|
||||
transport: odinTransport,
|
||||
receiptStore: durableReceiptStore,
|
||||
publications: [advertisementPublication, surfacePublication],
|
||||
commandHandlers: {
|
||||
"swarm.set_heat": applyHeatCommandTransaction,
|
||||
},
|
||||
});
|
||||
|
||||
await session.start();
|
||||
await session.upsertPublication(nextSurfacePublication);
|
||||
```
|
||||
|
||||
Provider, service-instance, endpoint, and body-producer identities are separate
|
||||
contracts. The provider session never infers one from another. A durable
|
||||
receipt store is required; `CultMeshMemoryProviderReceiptStore` is intended
|
||||
only for tests and disposable tools. The store is a durable outbox: a receipt
|
||||
is persisted before transport publication, remains pending across reconnects,
|
||||
and is marked published only after the current connection accepts it. Receipt
|
||||
store failure degrades command intake without inventing a network outage.
|
||||
|
||||
## RUDP Helpers
|
||||
|
||||
Node runtimes can build the shared CultNet reliable-UDP transport from the
|
||||
|
|
|
|||
|
|
@ -32,6 +32,8 @@ import {
|
|||
type CultNetWireContract,
|
||||
} from "cultnet-ts";
|
||||
|
||||
export * from "./provider-session";
|
||||
|
||||
export interface CultMeshVec2 {
|
||||
readonly x: number;
|
||||
readonly y: number;
|
||||
|
|
|
|||
884
packages/cultmesh-ts/src/provider-session.ts
Normal file
884
packages/cultmesh-ts/src/provider-session.ts
Normal file
|
|
@ -0,0 +1,884 @@
|
|||
export interface CultMeshProviderIdentity {
|
||||
readonly providerId: string;
|
||||
readonly serviceInstanceId: string;
|
||||
readonly endpointId: string;
|
||||
readonly verseId: string;
|
||||
}
|
||||
|
||||
export interface CultMeshProviderLease {
|
||||
readonly leaseId: string;
|
||||
readonly expiresAt: Date;
|
||||
}
|
||||
|
||||
export interface CultMeshProviderPublication<TValue = unknown> {
|
||||
readonly publicationId: string;
|
||||
readonly documentType: string;
|
||||
readonly schemaId: string;
|
||||
readonly recordKey: string;
|
||||
readonly value: TValue;
|
||||
}
|
||||
|
||||
export interface CultMeshProviderCommand<TPayload = unknown> {
|
||||
readonly commandId: string;
|
||||
readonly commandKind: string;
|
||||
readonly providerId: string;
|
||||
readonly serviceInstanceId: string;
|
||||
readonly payload: TPayload;
|
||||
}
|
||||
|
||||
export type CultMeshProviderReceiptState = "applied" | "rejected" | "failed";
|
||||
|
||||
export interface CultMeshProviderCommandReceipt<TResult = unknown> {
|
||||
readonly receiptId: string;
|
||||
readonly commandId: string;
|
||||
readonly commandKind: string;
|
||||
readonly providerId: string;
|
||||
readonly serviceInstanceId: string;
|
||||
readonly state: CultMeshProviderReceiptState;
|
||||
readonly completedAt: Date;
|
||||
readonly result?: TResult;
|
||||
readonly error?: string;
|
||||
}
|
||||
|
||||
export interface CultMeshProviderCommandOutcome<TResult = unknown> {
|
||||
readonly state: CultMeshProviderReceiptState;
|
||||
readonly result?: TResult;
|
||||
readonly error?: string;
|
||||
}
|
||||
|
||||
export interface CultMeshProviderCommandContext {
|
||||
readonly identity: CultMeshProviderIdentity;
|
||||
}
|
||||
|
||||
export type CultMeshProviderCommandHandler = (
|
||||
command: CultMeshProviderCommand,
|
||||
context: CultMeshProviderCommandContext,
|
||||
) => Promise<CultMeshProviderCommandOutcome>;
|
||||
|
||||
export interface CultMeshProviderReceiptStore {
|
||||
get(commandId: string): Promise<CultMeshProviderCommandReceipt | undefined>;
|
||||
put(receipt: CultMeshProviderCommandReceipt): Promise<void>;
|
||||
listPending(): Promise<readonly CultMeshProviderCommandReceipt[]>;
|
||||
markPublished(receiptId: string): Promise<void>;
|
||||
}
|
||||
|
||||
export class CultMeshMemoryProviderReceiptStore implements CultMeshProviderReceiptStore {
|
||||
readonly #receipts = new Map<string, CultMeshProviderCommandReceipt>();
|
||||
readonly #pendingReceiptIds = new Set<string>();
|
||||
|
||||
public async get(commandId: string): Promise<CultMeshProviderCommandReceipt | undefined> {
|
||||
return this.#receipts.get(commandId);
|
||||
}
|
||||
|
||||
public async put(receipt: CultMeshProviderCommandReceipt): Promise<void> {
|
||||
this.#receipts.set(receipt.commandId, receipt);
|
||||
this.#pendingReceiptIds.add(receipt.receiptId);
|
||||
}
|
||||
|
||||
public async listPending(): Promise<readonly CultMeshProviderCommandReceipt[]> {
|
||||
return [...this.#receipts.values()]
|
||||
.filter(receipt => this.#pendingReceiptIds.has(receipt.receiptId))
|
||||
.sort((left, right) => left.receiptId.localeCompare(right.receiptId));
|
||||
}
|
||||
|
||||
public async markPublished(receiptId: string): Promise<void> {
|
||||
this.#pendingReceiptIds.delete(receiptId);
|
||||
}
|
||||
}
|
||||
|
||||
export interface CultMeshProviderRegistration {
|
||||
readonly identity: CultMeshProviderIdentity;
|
||||
readonly requestedLeaseDurationMs: number;
|
||||
}
|
||||
|
||||
export interface CultMeshProviderWithdrawal {
|
||||
readonly identity: CultMeshProviderIdentity;
|
||||
readonly leaseId: string;
|
||||
readonly publicationIds: readonly string[];
|
||||
}
|
||||
|
||||
export type CultMeshProviderCommandListener = (
|
||||
command: CultMeshProviderCommand,
|
||||
) => void | Promise<void>;
|
||||
|
||||
export type CultMeshProviderUnsubscribe = () => void | Promise<void>;
|
||||
|
||||
export interface CultMeshProviderConnection {
|
||||
register(
|
||||
registration: CultMeshProviderRegistration,
|
||||
signal: AbortSignal,
|
||||
): Promise<CultMeshProviderLease>;
|
||||
renew(lease: CultMeshProviderLease): Promise<CultMeshProviderLease>;
|
||||
publish(
|
||||
publication: CultMeshProviderPublication,
|
||||
lease: CultMeshProviderLease,
|
||||
): Promise<void>;
|
||||
withdrawPublication(publicationId: string, lease: CultMeshProviderLease): Promise<void>;
|
||||
watchCommands(listener: CultMeshProviderCommandListener): CultMeshProviderUnsubscribe;
|
||||
publishReceipt(
|
||||
receipt: CultMeshProviderCommandReceipt,
|
||||
lease: CultMeshProviderLease,
|
||||
): Promise<void>;
|
||||
withdraw(withdrawal: CultMeshProviderWithdrawal): Promise<void>;
|
||||
close(): void | Promise<void>;
|
||||
}
|
||||
|
||||
export interface CultMeshProviderTransport {
|
||||
connect(
|
||||
identity: CultMeshProviderIdentity,
|
||||
signal: AbortSignal,
|
||||
): Promise<CultMeshProviderConnection>;
|
||||
}
|
||||
|
||||
export interface CultMeshProviderScheduledTask {
|
||||
cancel(): void;
|
||||
}
|
||||
|
||||
export interface CultMeshProviderScheduler {
|
||||
now(): Date;
|
||||
schedule(delayMs: number, action: () => void | Promise<void>): CultMeshProviderScheduledTask;
|
||||
}
|
||||
|
||||
export class CultMeshSystemProviderScheduler implements CultMeshProviderScheduler {
|
||||
public now(): Date {
|
||||
return new Date();
|
||||
}
|
||||
|
||||
public schedule(delayMs: number, action: () => void | Promise<void>): CultMeshProviderScheduledTask {
|
||||
const timer = setTimeout(() => {
|
||||
void Promise.resolve(action()).catch(() => undefined);
|
||||
}, Math.max(0, delayMs));
|
||||
return { cancel: () => clearTimeout(timer) };
|
||||
}
|
||||
}
|
||||
|
||||
export type CultMeshProviderSessionStatus =
|
||||
| "stopped"
|
||||
| "connecting"
|
||||
| "active"
|
||||
| "degraded"
|
||||
| "reconnecting"
|
||||
| "withdrawing";
|
||||
|
||||
export interface CultMeshProviderSessionState {
|
||||
readonly status: CultMeshProviderSessionStatus;
|
||||
readonly identity: CultMeshProviderIdentity;
|
||||
readonly reconnectAttempt: number;
|
||||
readonly leaseId?: string;
|
||||
readonly leaseExpiresAt?: Date;
|
||||
readonly lastError?: string;
|
||||
}
|
||||
|
||||
export interface CultMeshProviderSessionOptions {
|
||||
readonly identity: CultMeshProviderIdentity;
|
||||
readonly transport: CultMeshProviderTransport;
|
||||
readonly publications?: Iterable<CultMeshProviderPublication>;
|
||||
readonly commandHandlers?: Readonly<Record<string, CultMeshProviderCommandHandler>>;
|
||||
readonly receiptStore: CultMeshProviderReceiptStore;
|
||||
readonly scheduler?: CultMeshProviderScheduler;
|
||||
readonly onObserverError?: (error: unknown) => void;
|
||||
readonly leaseDurationMs?: number;
|
||||
readonly renewalLeadMs?: number;
|
||||
readonly reconnectBaseDelayMs?: number;
|
||||
readonly reconnectMaxDelayMs?: number;
|
||||
}
|
||||
|
||||
export class CultMeshProviderSession {
|
||||
readonly #identity: CultMeshProviderIdentity;
|
||||
readonly #transport: CultMeshProviderTransport;
|
||||
readonly #receiptStore: CultMeshProviderReceiptStore;
|
||||
readonly #scheduler: CultMeshProviderScheduler;
|
||||
readonly #onObserverError: (error: unknown) => void;
|
||||
readonly #leaseDurationMs: number;
|
||||
readonly #renewalLeadMs: number;
|
||||
readonly #reconnectBaseDelayMs: number;
|
||||
readonly #reconnectMaxDelayMs: number;
|
||||
readonly #publications = new Map<string, CultMeshProviderPublication>();
|
||||
readonly #handlers = new Map<string, CultMeshProviderCommandHandler>();
|
||||
readonly #subscribers = new Set<(state: CultMeshProviderSessionState) => void>();
|
||||
readonly #inFlightCommands = new Map<string, {
|
||||
command: CultMeshProviderCommand;
|
||||
work: Promise<void>;
|
||||
}>();
|
||||
#state: CultMeshProviderSessionState;
|
||||
#connection?: CultMeshProviderConnection;
|
||||
#lease?: CultMeshProviderLease;
|
||||
#unsubscribeCommands?: CultMeshProviderUnsubscribe;
|
||||
#scheduled?: CultMeshProviderScheduledTask;
|
||||
#abort?: AbortController;
|
||||
#running = false;
|
||||
#generation = 0;
|
||||
#reconnectAttempt = 0;
|
||||
#publicationWork: Promise<void> = Promise.resolve();
|
||||
#startWork?: Promise<void>;
|
||||
#stopWork?: Promise<void>;
|
||||
#lossWork?: { generation: number; work: Promise<void> };
|
||||
|
||||
public constructor(options: CultMeshProviderSessionOptions) {
|
||||
this.#identity = validateIdentity(options.identity);
|
||||
this.#transport = options.transport;
|
||||
if (!options.receiptStore) {
|
||||
throw new Error(
|
||||
"CultMesh provider sessions require an explicit receiptStore so command receipts survive the provider lifecycle.",
|
||||
);
|
||||
}
|
||||
this.#receiptStore = options.receiptStore;
|
||||
this.#scheduler = options.scheduler ?? new CultMeshSystemProviderScheduler();
|
||||
this.#onObserverError = options.onObserverError ?? (() => undefined);
|
||||
this.#leaseDurationMs = positive(options.leaseDurationMs ?? 30_000, "leaseDurationMs");
|
||||
this.#renewalLeadMs = nonNegative(options.renewalLeadMs ?? 10_000, "renewalLeadMs");
|
||||
if (this.#renewalLeadMs >= this.#leaseDurationMs) {
|
||||
throw new Error("CultMesh provider renewalLeadMs must be less than leaseDurationMs.");
|
||||
}
|
||||
this.#reconnectBaseDelayMs = positive(
|
||||
options.reconnectBaseDelayMs ?? 250,
|
||||
"reconnectBaseDelayMs",
|
||||
);
|
||||
this.#reconnectMaxDelayMs = positive(
|
||||
options.reconnectMaxDelayMs ?? 10_000,
|
||||
"reconnectMaxDelayMs",
|
||||
);
|
||||
if (this.#reconnectMaxDelayMs < this.#reconnectBaseDelayMs) {
|
||||
throw new Error("CultMesh provider reconnectMaxDelayMs must not be less than reconnectBaseDelayMs.");
|
||||
}
|
||||
for (const publication of options.publications ?? []) {
|
||||
this.#publications.set(validatePublication(publication).publicationId, publication);
|
||||
}
|
||||
for (const [kind, handler] of Object.entries(options.commandHandlers ?? {})) {
|
||||
this.setCommandHandler(kind, handler);
|
||||
}
|
||||
this.#state = {
|
||||
status: "stopped",
|
||||
identity: this.#identity,
|
||||
reconnectAttempt: 0,
|
||||
};
|
||||
}
|
||||
|
||||
public get identity(): CultMeshProviderIdentity {
|
||||
return this.#identity;
|
||||
}
|
||||
|
||||
public get state(): CultMeshProviderSessionState {
|
||||
return this.#state;
|
||||
}
|
||||
|
||||
public get publications(): readonly CultMeshProviderPublication[] {
|
||||
return [...this.#publications.values()].sort((left, right) =>
|
||||
left.publicationId.localeCompare(right.publicationId),
|
||||
);
|
||||
}
|
||||
|
||||
public watchState(callback: (state: CultMeshProviderSessionState) => void): () => void {
|
||||
this.#subscribers.add(callback);
|
||||
try {
|
||||
callback(this.#state);
|
||||
} catch (error) {
|
||||
this.#reportObserverError(error);
|
||||
}
|
||||
return () => this.#subscribers.delete(callback);
|
||||
}
|
||||
|
||||
public setCommandHandler(kind: string, handler: CultMeshProviderCommandHandler): void {
|
||||
this.#handlers.set(required(kind, "command kind"), handler);
|
||||
}
|
||||
|
||||
public removeCommandHandler(kind: string): boolean {
|
||||
return this.#handlers.delete(required(kind, "command kind"));
|
||||
}
|
||||
|
||||
public upsertPublication(publication: CultMeshProviderPublication): Promise<void> {
|
||||
const valid = validatePublication(publication);
|
||||
this.#publications.set(valid.publicationId, valid);
|
||||
if (this.#connection && this.#lease) {
|
||||
return this.#enqueuePublication(() => this.#deliverPublication(valid, this.#generation));
|
||||
}
|
||||
return Promise.resolve();
|
||||
}
|
||||
|
||||
public async removePublication(publicationId: string): Promise<boolean> {
|
||||
const id = required(publicationId, "publicationId");
|
||||
const removed = this.#publications.delete(id);
|
||||
if (removed && this.#connection && this.#lease) {
|
||||
await this.#enqueuePublication(() => this.#withdrawPublication(id, this.#generation));
|
||||
}
|
||||
return removed;
|
||||
}
|
||||
|
||||
public start(): Promise<void> {
|
||||
if (this.#stopWork) {
|
||||
return Promise.reject(new Error("CultMesh provider cannot start while shutdown is in progress."));
|
||||
}
|
||||
if (this.#startWork) return this.#startWork;
|
||||
if (this.#running) return Promise.resolve();
|
||||
let tracked!: Promise<void>;
|
||||
tracked = Promise.resolve().then(() => this.#start()).finally(() => {
|
||||
if (this.#startWork === tracked) this.#startWork = undefined;
|
||||
});
|
||||
this.#startWork = tracked;
|
||||
return tracked;
|
||||
}
|
||||
|
||||
async #start(): Promise<void> {
|
||||
this.#running = true;
|
||||
this.#reconnectAttempt = 0;
|
||||
const generation = ++this.#generation;
|
||||
this.#abort = new AbortController();
|
||||
await this.#connect(generation, false);
|
||||
}
|
||||
|
||||
public stop(): Promise<void> {
|
||||
if (this.#stopWork) return this.#stopWork;
|
||||
if (!this.#running && !this.#startWork && this.#state.status === "stopped") {
|
||||
return Promise.resolve();
|
||||
}
|
||||
let tracked!: Promise<void>;
|
||||
tracked = Promise.resolve().then(() => this.#stop()).finally(() => {
|
||||
if (this.#stopWork === tracked) this.#stopWork = undefined;
|
||||
});
|
||||
this.#stopWork = tracked;
|
||||
return tracked;
|
||||
}
|
||||
|
||||
async #stop(): Promise<void> {
|
||||
this.#running = false;
|
||||
const generation = ++this.#generation;
|
||||
this.#cancelScheduled();
|
||||
this.#abort?.abort();
|
||||
this.#setState("withdrawing");
|
||||
const connection = this.#connection;
|
||||
const lease = this.#lease;
|
||||
const starting = this.#startWork;
|
||||
const losing = this.#lossWork?.work;
|
||||
this.#connection = undefined;
|
||||
this.#lease = undefined;
|
||||
const errors: unknown[] = [];
|
||||
try {
|
||||
await this.#unsubscribe();
|
||||
} catch (error) {
|
||||
errors.push(error);
|
||||
}
|
||||
try {
|
||||
await this.#publicationWork;
|
||||
} catch (error) {
|
||||
errors.push(error);
|
||||
}
|
||||
for (const lifecycleWork of [starting, losing]) {
|
||||
if (!lifecycleWork) continue;
|
||||
try {
|
||||
await lifecycleWork;
|
||||
} catch (error) {
|
||||
errors.push(error);
|
||||
}
|
||||
}
|
||||
this.#cancelScheduled();
|
||||
const commandWork = [...this.#inFlightCommands.values()].map(value => value.work);
|
||||
const commandResults = await Promise.allSettled(commandWork);
|
||||
for (const result of commandResults) {
|
||||
if (result.status === "rejected") errors.push(result.reason);
|
||||
}
|
||||
if (connection && lease) {
|
||||
try {
|
||||
await connection.withdraw({
|
||||
identity: this.#identity,
|
||||
leaseId: lease.leaseId,
|
||||
publicationIds: this.publications.map(value => value.publicationId),
|
||||
});
|
||||
} catch (error) {
|
||||
errors.push(error);
|
||||
}
|
||||
}
|
||||
if (connection) {
|
||||
try {
|
||||
await connection.close();
|
||||
} catch (error) {
|
||||
errors.push(error);
|
||||
}
|
||||
}
|
||||
if (generation === this.#generation) this.#setState("stopped");
|
||||
if (errors.length > 0) {
|
||||
throw new AggregateError(errors, "CultMesh provider shutdown completed with cleanup failures.");
|
||||
}
|
||||
}
|
||||
|
||||
async #connect(generation: number, reconnecting: boolean): Promise<void> {
|
||||
if (!this.#isCurrent(generation)) return;
|
||||
this.#scheduled = undefined;
|
||||
this.#setState(reconnecting ? "reconnecting" : "connecting");
|
||||
try {
|
||||
const connection = await this.#transport.connect(this.#identity, this.#abort!.signal);
|
||||
if (!this.#isCurrent(generation)) {
|
||||
await connection.close();
|
||||
return;
|
||||
}
|
||||
this.#connection = connection;
|
||||
const lease = validateLease(
|
||||
await connection.register({
|
||||
identity: this.#identity,
|
||||
requestedLeaseDurationMs: this.#leaseDurationMs,
|
||||
}, this.#abort!.signal),
|
||||
this.#scheduler.now(),
|
||||
);
|
||||
if (!this.#isCurrent(generation)) {
|
||||
await connection.close();
|
||||
return;
|
||||
}
|
||||
this.#lease = lease;
|
||||
const initialPublications = this.publications;
|
||||
await this.#enqueuePublication(async () => {
|
||||
for (const publication of initialPublications) {
|
||||
if (!this.#isCurrent(generation)) return;
|
||||
requireUnexpiredLease(lease, this.#scheduler.now());
|
||||
await connection.publish(publication, lease);
|
||||
}
|
||||
});
|
||||
if (!this.#isCurrent(generation)) return;
|
||||
const readyLease = await this.#renewIfNeeded(connection, lease, generation);
|
||||
if (!readyLease || !this.#isCurrent(generation)) return;
|
||||
this.#lease = readyLease;
|
||||
this.#reconnectAttempt = 0;
|
||||
const receiptsReady = await this.#drainPendingReceipts(connection, readyLease, generation);
|
||||
if (!this.#isCurrent(generation)) return;
|
||||
if (receiptsReady) {
|
||||
this.#enableCommands(connection, generation);
|
||||
this.#setState("active", undefined, readyLease);
|
||||
}
|
||||
this.#scheduleRenewal(generation, readyLease);
|
||||
} catch (error) {
|
||||
await this.#loseConnection(generation, error);
|
||||
}
|
||||
}
|
||||
|
||||
#scheduleRenewal(generation: number, lease: CultMeshProviderLease): void {
|
||||
if (!this.#isCurrent(generation)) return;
|
||||
this.#scheduled?.cancel();
|
||||
const delay = Math.max(
|
||||
1,
|
||||
lease.expiresAt.getTime() - this.#scheduler.now().getTime() - this.#renewalLeadMs,
|
||||
);
|
||||
this.#scheduled = this.#scheduler.schedule(delay, () => this.#renew(generation));
|
||||
}
|
||||
|
||||
async #renew(generation: number): Promise<void> {
|
||||
if (!this.#isCurrent(generation) || !this.#connection || !this.#lease) return;
|
||||
this.#scheduled = undefined;
|
||||
try {
|
||||
let lease = validateLease(
|
||||
await this.#connection.renew(this.#lease),
|
||||
this.#scheduler.now(),
|
||||
);
|
||||
if (!this.#isCurrent(generation)) return;
|
||||
this.#lease = lease;
|
||||
const receiptsReady = await this.#drainPendingReceipts(this.#connection, lease, generation);
|
||||
if (!this.#isCurrent(generation)) return;
|
||||
if (receiptsReady) {
|
||||
lease = await this.#renewIfNeeded(this.#connection, lease, generation) ?? lease;
|
||||
if (!this.#isCurrent(generation)) return;
|
||||
this.#lease = lease;
|
||||
this.#enableCommands(this.#connection, generation);
|
||||
this.#setState("active", undefined, lease);
|
||||
}
|
||||
this.#scheduleRenewal(generation, lease);
|
||||
} catch (error) {
|
||||
await this.#loseConnection(generation, error);
|
||||
}
|
||||
}
|
||||
|
||||
async #deliverPublication(
|
||||
publication: CultMeshProviderPublication,
|
||||
generation: number,
|
||||
): Promise<void> {
|
||||
const connection = this.#connection;
|
||||
const lease = this.#lease;
|
||||
if (!connection || !lease || !this.#isCurrent(generation)) return;
|
||||
try {
|
||||
requireUnexpiredLease(lease, this.#scheduler.now());
|
||||
await connection.publish(publication, lease);
|
||||
} catch (error) {
|
||||
await this.#loseConnection(generation, error);
|
||||
}
|
||||
}
|
||||
|
||||
async #withdrawPublication(publicationId: string, generation: number): Promise<void> {
|
||||
const connection = this.#connection;
|
||||
const lease = this.#lease;
|
||||
if (!connection || !lease || !this.#isCurrent(generation)) return;
|
||||
try {
|
||||
requireUnexpiredLease(lease, this.#scheduler.now());
|
||||
await connection.withdrawPublication(publicationId, lease);
|
||||
} catch (error) {
|
||||
await this.#loseConnection(generation, error);
|
||||
}
|
||||
}
|
||||
|
||||
async #receiveCommand(command: CultMeshProviderCommand, generation: number): Promise<void> {
|
||||
if (!this.#isCurrent(generation) ||
|
||||
command.providerId !== this.#identity.providerId ||
|
||||
command.serviceInstanceId !== this.#identity.serviceInstanceId) {
|
||||
return;
|
||||
}
|
||||
try {
|
||||
required(command.commandId, "commandId");
|
||||
required(command.commandKind, "commandKind");
|
||||
} catch (error) {
|
||||
this.#setState("degraded", errorText(error), this.#lease);
|
||||
return;
|
||||
}
|
||||
const inFlight = this.#inFlightCommands.get(command.commandId);
|
||||
if (inFlight) {
|
||||
if (!sameCommandIdentity(inFlight.command, command)) {
|
||||
this.#setState(
|
||||
"degraded",
|
||||
`CultMesh command id '${command.commandId}' was reused with different identity while in flight.`,
|
||||
this.#lease,
|
||||
);
|
||||
return;
|
||||
}
|
||||
await inFlight.work;
|
||||
return;
|
||||
}
|
||||
const work = this.#processCommand(command, generation);
|
||||
const entry = { command, work };
|
||||
this.#inFlightCommands.set(command.commandId, entry);
|
||||
try {
|
||||
await work;
|
||||
} finally {
|
||||
if (this.#inFlightCommands.get(command.commandId) === entry) {
|
||||
this.#inFlightCommands.delete(command.commandId);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
async #processCommand(command: CultMeshProviderCommand, generation: number): Promise<void> {
|
||||
let receipt: CultMeshProviderCommandReceipt;
|
||||
try {
|
||||
receipt = await this.#resolveCommand(command);
|
||||
} catch (error) {
|
||||
if (this.#isCurrent(generation)) {
|
||||
await this.#degradeReceiptStore(
|
||||
`Command receipt persistence failed: ${errorText(error)}`,
|
||||
generation,
|
||||
);
|
||||
}
|
||||
return;
|
||||
}
|
||||
const connection = this.#connection;
|
||||
const lease = this.#lease;
|
||||
if (!connection || !lease || !this.#isCurrent(generation)) return;
|
||||
try {
|
||||
await connection.publishReceipt(receipt, lease);
|
||||
} catch (error) {
|
||||
await this.#loseConnection(generation, error);
|
||||
return;
|
||||
}
|
||||
try {
|
||||
await this.#receiptStore.markPublished(receipt.receiptId);
|
||||
} catch (error) {
|
||||
if (this.#isCurrent(generation)) {
|
||||
await this.#degradeReceiptStore(
|
||||
`Command receipt publication could not be committed: ${errorText(error)}`,
|
||||
generation,
|
||||
);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
async #resolveCommand(command: CultMeshProviderCommand): Promise<CultMeshProviderCommandReceipt> {
|
||||
const prior = await this.#receiptStore.get(command.commandId);
|
||||
if (prior) {
|
||||
validateStoredReceipt(prior, this.#identity);
|
||||
if (prior.commandKind !== command.commandKind) {
|
||||
throw new Error(`CultMesh command id '${command.commandId}' was reused with different identity.`);
|
||||
}
|
||||
return prior;
|
||||
}
|
||||
const handler = this.#handlers.get(command.commandKind);
|
||||
let outcome: CultMeshProviderCommandOutcome;
|
||||
if (!handler) {
|
||||
outcome = { state: "rejected", error: `No handler for '${command.commandKind}'.` };
|
||||
} else {
|
||||
try {
|
||||
outcome = await handler(command, { identity: this.#identity });
|
||||
validateOutcomeState(outcome?.state);
|
||||
} catch (error) {
|
||||
outcome = { state: "failed", error: errorText(error) };
|
||||
}
|
||||
}
|
||||
const receipt: CultMeshProviderCommandReceipt = {
|
||||
receiptId: receiptId(this.#identity, command.commandId),
|
||||
commandId: command.commandId,
|
||||
commandKind: command.commandKind,
|
||||
providerId: this.#identity.providerId,
|
||||
serviceInstanceId: this.#identity.serviceInstanceId,
|
||||
state: validateOutcomeState(outcome.state),
|
||||
completedAt: this.#scheduler.now(),
|
||||
result: outcome.result,
|
||||
error: outcome.error,
|
||||
};
|
||||
await this.#receiptStore.put(receipt);
|
||||
return receipt;
|
||||
}
|
||||
|
||||
async #loseConnection(generation: number, error: unknown): Promise<void> {
|
||||
if (!this.#isCurrent(generation)) return;
|
||||
const existing = this.#lossWork;
|
||||
if (existing?.generation === generation) {
|
||||
await existing.work;
|
||||
return;
|
||||
}
|
||||
let tracked!: Promise<void>;
|
||||
tracked = this.#performConnectionLoss(generation, error).finally(() => {
|
||||
if (this.#lossWork?.work === tracked) this.#lossWork = undefined;
|
||||
});
|
||||
this.#lossWork = { generation, work: tracked };
|
||||
await tracked;
|
||||
}
|
||||
|
||||
async #performConnectionLoss(generation: number, error: unknown): Promise<void> {
|
||||
this.#cancelScheduled();
|
||||
const connection = this.#connection;
|
||||
this.#connection = undefined;
|
||||
this.#lease = undefined;
|
||||
this.#reconnectAttempt++;
|
||||
const failures = [errorText(error)];
|
||||
this.#setState("reconnecting", failures[0]);
|
||||
try {
|
||||
await this.#unsubscribe();
|
||||
} catch (cleanupError) {
|
||||
failures.push(`unsubscribe failed: ${errorText(cleanupError)}`);
|
||||
}
|
||||
if (connection) {
|
||||
try {
|
||||
await connection.close();
|
||||
} catch (cleanupError) {
|
||||
failures.push(`close failed: ${errorText(cleanupError)}`);
|
||||
}
|
||||
}
|
||||
if (!this.#isCurrent(generation)) return;
|
||||
this.#setState("reconnecting", failures.join("; "));
|
||||
const exponent = Math.max(0, this.#reconnectAttempt - 1);
|
||||
const delay = Math.min(
|
||||
this.#reconnectMaxDelayMs,
|
||||
this.#reconnectBaseDelayMs * 2 ** exponent,
|
||||
);
|
||||
this.#scheduled = this.#scheduler.schedule(delay, () => this.#connect(generation, true));
|
||||
}
|
||||
|
||||
async #drainPendingReceipts(
|
||||
connection: CultMeshProviderConnection,
|
||||
lease: CultMeshProviderLease,
|
||||
generation: number,
|
||||
): Promise<boolean> {
|
||||
let pending: readonly CultMeshProviderCommandReceipt[];
|
||||
try {
|
||||
pending = await this.#receiptStore.listPending();
|
||||
} catch (error) {
|
||||
if (this.#isCurrent(generation)) {
|
||||
await this.#degradeReceiptStore(
|
||||
`Pending command receipts could not be read: ${errorText(error)}`,
|
||||
generation,
|
||||
);
|
||||
}
|
||||
return false;
|
||||
}
|
||||
for (const receipt of [...pending].sort((left, right) => left.receiptId.localeCompare(right.receiptId))) {
|
||||
try {
|
||||
validateStoredReceipt(receipt, this.#identity);
|
||||
} catch (error) {
|
||||
if (this.#isCurrent(generation)) {
|
||||
await this.#degradeReceiptStore(
|
||||
`Pending command receipt is invalid: ${errorText(error)}`,
|
||||
generation,
|
||||
);
|
||||
}
|
||||
return false;
|
||||
}
|
||||
if (!this.#isCurrent(generation)) return false;
|
||||
requireUnexpiredLease(lease, this.#scheduler.now());
|
||||
await connection.publishReceipt(receipt, lease);
|
||||
try {
|
||||
await this.#receiptStore.markPublished(receipt.receiptId);
|
||||
} catch (error) {
|
||||
if (this.#isCurrent(generation)) {
|
||||
await this.#degradeReceiptStore(
|
||||
`Command receipt publication could not be committed: ${errorText(error)}`,
|
||||
generation,
|
||||
);
|
||||
}
|
||||
return false;
|
||||
}
|
||||
}
|
||||
return true;
|
||||
}
|
||||
|
||||
async #renewIfNeeded(
|
||||
connection: CultMeshProviderConnection,
|
||||
lease: CultMeshProviderLease,
|
||||
generation: number,
|
||||
): Promise<CultMeshProviderLease | undefined> {
|
||||
if (lease.expiresAt.getTime() - this.#scheduler.now().getTime() > this.#renewalLeadMs) {
|
||||
return lease;
|
||||
}
|
||||
const renewed = validateLease(await connection.renew(lease), this.#scheduler.now());
|
||||
return this.#isCurrent(generation) ? renewed : undefined;
|
||||
}
|
||||
|
||||
#enableCommands(connection: CultMeshProviderConnection, generation: number): void {
|
||||
if (this.#unsubscribeCommands || !this.#isCurrent(generation)) return;
|
||||
this.#unsubscribeCommands = connection.watchCommands(command => this.#receiveCommand(command, generation));
|
||||
}
|
||||
|
||||
async #degradeReceiptStore(message: string, generation: number): Promise<void> {
|
||||
try {
|
||||
await this.#unsubscribe();
|
||||
} catch (error) {
|
||||
this.#reportObserverError(error);
|
||||
}
|
||||
if (this.#isCurrent(generation)) this.#setState("degraded", message, this.#lease);
|
||||
}
|
||||
|
||||
async #unsubscribe(): Promise<void> {
|
||||
const unsubscribe = this.#unsubscribeCommands;
|
||||
this.#unsubscribeCommands = undefined;
|
||||
if (unsubscribe) await unsubscribe();
|
||||
}
|
||||
|
||||
#cancelScheduled(): void {
|
||||
const scheduled = this.#scheduled;
|
||||
this.#scheduled = undefined;
|
||||
scheduled?.cancel();
|
||||
}
|
||||
|
||||
#enqueuePublication(action: () => Promise<void>): Promise<void> {
|
||||
const work = this.#publicationWork.then(action, action);
|
||||
this.#publicationWork = work.catch(() => undefined);
|
||||
return work;
|
||||
}
|
||||
|
||||
#isCurrent(generation: number): boolean {
|
||||
return this.#running && generation === this.#generation;
|
||||
}
|
||||
|
||||
#setState(
|
||||
status: CultMeshProviderSessionStatus,
|
||||
lastError?: string,
|
||||
lease?: CultMeshProviderLease,
|
||||
): void {
|
||||
this.#state = {
|
||||
status,
|
||||
identity: this.#identity,
|
||||
reconnectAttempt: this.#reconnectAttempt,
|
||||
leaseId: lease?.leaseId,
|
||||
leaseExpiresAt: lease?.expiresAt,
|
||||
lastError,
|
||||
};
|
||||
for (const subscriber of [...this.#subscribers]) {
|
||||
try {
|
||||
subscriber(this.#state);
|
||||
} catch (error) {
|
||||
this.#reportObserverError(error);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#reportObserverError(error: unknown): void {
|
||||
try {
|
||||
this.#onObserverError(error);
|
||||
} catch {
|
||||
// Diagnostics cannot become a provider lifecycle authority.
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
function validateIdentity(identity: CultMeshProviderIdentity): CultMeshProviderIdentity {
|
||||
required(identity.providerId, "providerId");
|
||||
required(identity.serviceInstanceId, "serviceInstanceId");
|
||||
required(identity.endpointId, "endpointId");
|
||||
required(identity.verseId, "verseId");
|
||||
return identity;
|
||||
}
|
||||
|
||||
function validatePublication(publication: CultMeshProviderPublication): CultMeshProviderPublication {
|
||||
required(publication.publicationId, "publicationId");
|
||||
required(publication.documentType, "documentType");
|
||||
required(publication.schemaId, "schemaId");
|
||||
required(publication.recordKey, "recordKey");
|
||||
return publication;
|
||||
}
|
||||
|
||||
function validateLease(lease: CultMeshProviderLease, now: Date): CultMeshProviderLease {
|
||||
required(lease.leaseId, "leaseId");
|
||||
if (!(lease.expiresAt instanceof Date) || !Number.isFinite(lease.expiresAt.getTime())) {
|
||||
throw new Error("CultMesh provider lease requires a valid expiresAt date.");
|
||||
}
|
||||
if (lease.expiresAt <= now) {
|
||||
throw new Error("CultMesh provider lease must expire in the future.");
|
||||
}
|
||||
return lease;
|
||||
}
|
||||
|
||||
function requireUnexpiredLease(lease: CultMeshProviderLease, now: Date): void {
|
||||
if (lease.expiresAt <= now) {
|
||||
throw new Error(`CultMesh provider lease '${lease.leaseId}' expired during replay.`);
|
||||
}
|
||||
}
|
||||
|
||||
function validateOutcomeState(state: CultMeshProviderReceiptState): CultMeshProviderReceiptState {
|
||||
if (state !== "applied" && state !== "rejected" && state !== "failed") {
|
||||
throw new Error(`CultMesh provider command handler returned invalid receipt state '${String(state)}'.`);
|
||||
}
|
||||
return state;
|
||||
}
|
||||
|
||||
function sameCommandIdentity(
|
||||
left: CultMeshProviderCommand,
|
||||
right: CultMeshProviderCommand,
|
||||
): boolean {
|
||||
return left.providerId === right.providerId &&
|
||||
left.serviceInstanceId === right.serviceInstanceId &&
|
||||
left.commandKind === right.commandKind;
|
||||
}
|
||||
|
||||
function validateStoredReceipt(
|
||||
receipt: CultMeshProviderCommandReceipt,
|
||||
identity: CultMeshProviderIdentity,
|
||||
): void {
|
||||
required(receipt.commandId, "stored receipt commandId");
|
||||
required(receipt.commandKind, "stored receipt commandKind");
|
||||
required(receipt.receiptId, "stored receipt receiptId");
|
||||
validateOutcomeState(receipt.state);
|
||||
if (receipt.providerId !== identity.providerId ||
|
||||
receipt.serviceInstanceId !== identity.serviceInstanceId ||
|
||||
receipt.receiptId !== receiptId(identity, receipt.commandId)) {
|
||||
throw new Error(`CultMesh stored receipt '${receipt.receiptId}' does not belong to this provider session.`);
|
||||
}
|
||||
}
|
||||
|
||||
function receiptId(identity: CultMeshProviderIdentity, commandId: string): string {
|
||||
const parts = [identity.providerId, identity.serviceInstanceId, commandId];
|
||||
return `receipt:${parts.map(value => `${value.length}:${value}`).join("")}`;
|
||||
}
|
||||
|
||||
function required(value: string, name: string): string {
|
||||
if (!value || value.trim().length === 0) {
|
||||
throw new Error(`CultMesh provider ${name} must be non-empty.`);
|
||||
}
|
||||
return value;
|
||||
}
|
||||
|
||||
function positive(value: number, name: string): number {
|
||||
if (!Number.isFinite(value) || value <= 0) {
|
||||
throw new Error(`CultMesh provider ${name} must be positive.`);
|
||||
}
|
||||
return value;
|
||||
}
|
||||
|
||||
function nonNegative(value: number, name: string): number {
|
||||
if (!Number.isFinite(value) || value < 0) {
|
||||
throw new Error(`CultMesh provider ${name} must be non-negative.`);
|
||||
}
|
||||
return value;
|
||||
}
|
||||
|
||||
function errorText(error: unknown): string {
|
||||
return error instanceof Error ? error.message : String(error);
|
||||
}
|
||||
791
packages/cultmesh-ts/test/provider-session.test.ts
Normal file
791
packages/cultmesh-ts/test/provider-session.test.ts
Normal file
|
|
@ -0,0 +1,791 @@
|
|||
import test from "node:test";
|
||||
import assert from "node:assert/strict";
|
||||
import {
|
||||
CultMeshMemoryProviderReceiptStore,
|
||||
CultMeshProviderSession,
|
||||
type CultMeshProviderCommand,
|
||||
type CultMeshProviderCommandListener,
|
||||
type CultMeshProviderCommandReceipt,
|
||||
type CultMeshProviderConnection,
|
||||
type CultMeshProviderIdentity,
|
||||
type CultMeshProviderLease,
|
||||
type CultMeshProviderPublication,
|
||||
type CultMeshProviderRegistration,
|
||||
type CultMeshProviderScheduledTask,
|
||||
type CultMeshProviderScheduler,
|
||||
type CultMeshProviderTransport,
|
||||
type CultMeshProviderWithdrawal,
|
||||
type CultMeshProviderUnsubscribe,
|
||||
} from "../src/provider-session";
|
||||
|
||||
const identity: CultMeshProviderIdentity = {
|
||||
providerId: "voidbot.swarm",
|
||||
serviceInstanceId: "voidbot-worker-7",
|
||||
endpointId: "odin:voidbot-worker-7",
|
||||
verseId: "voidbot.local",
|
||||
};
|
||||
|
||||
const surface: CultMeshProviderPublication = {
|
||||
publicationId: "voidbot.swarm.surface",
|
||||
documentType: "gamecult.eve.surface_state",
|
||||
schemaId: "gamecult.eve.surface_state.v1",
|
||||
recordKey: "voidbot.swarm",
|
||||
value: { version: 1 },
|
||||
};
|
||||
|
||||
const advertisement: CultMeshProviderPublication = {
|
||||
publicationId: "voidbot.swarm.advertisement",
|
||||
documentType: "gamecult.eve.provider_advertisement",
|
||||
schemaId: "gamecult.eve.provider_advertisement.v1",
|
||||
recordKey: "voidbot.swarm",
|
||||
value: { providerId: "voidbot.swarm" },
|
||||
};
|
||||
|
||||
test("provider session owns registration, renewal, desired publication, and withdrawal", async () => {
|
||||
const scheduler = new ManualScheduler();
|
||||
const transport = new FakeTransport(scheduler);
|
||||
const states: string[] = [];
|
||||
const session = new CultMeshProviderSession({
|
||||
identity,
|
||||
transport,
|
||||
scheduler,
|
||||
receiptStore: new CultMeshMemoryProviderReceiptStore(),
|
||||
publications: [surface, advertisement],
|
||||
leaseDurationMs: 1_000,
|
||||
renewalLeadMs: 200,
|
||||
reconnectBaseDelayMs: 100,
|
||||
reconnectMaxDelayMs: 1_000,
|
||||
});
|
||||
session.watchState(state => states.push(state.status));
|
||||
|
||||
await session.start();
|
||||
|
||||
const connection = transport.connections[0]!;
|
||||
assert.equal(session.state.status, "active");
|
||||
assert.deepEqual(connection.registrations[0]?.identity, identity);
|
||||
assert.deepEqual(
|
||||
connection.publications.map(value => value.publicationId),
|
||||
[advertisement.publicationId, surface.publicationId],
|
||||
"initial desired state is republished in stable order",
|
||||
);
|
||||
assert.notEqual(identity.providerId, identity.serviceInstanceId);
|
||||
assert.notEqual(identity.serviceInstanceId, identity.endpointId);
|
||||
|
||||
await scheduler.advance(800);
|
||||
assert.equal(connection.renewals.length, 1);
|
||||
assert.equal(session.state.status, "active");
|
||||
|
||||
const updatedSurface = { ...surface, value: { version: 2 } };
|
||||
const finalSurface = { ...surface, value: { version: 3 } };
|
||||
const publishBarrier = new Deferred<void>();
|
||||
connection.publishBarrier = publishBarrier;
|
||||
const updating = session.upsertPublication(updatedSurface);
|
||||
const finalizing = session.upsertPublication(finalSurface);
|
||||
await settle();
|
||||
assert.deepEqual(connection.publications.at(-1), surface, "the second update waits behind the first delivery");
|
||||
publishBarrier.resolve();
|
||||
await Promise.all([updating, finalizing]);
|
||||
assert.deepEqual(
|
||||
connection.publications.slice(-2),
|
||||
[updatedSurface, finalSurface],
|
||||
"desired publication updates preserve caller order on the live connection",
|
||||
);
|
||||
|
||||
assert.equal(await session.removePublication(surface.publicationId), true);
|
||||
assert.deepEqual(connection.withdrawnPublications, [surface.publicationId]);
|
||||
|
||||
await session.stop();
|
||||
assert.equal(connection.closed, true);
|
||||
assert.deepEqual(connection.withdrawals[0]?.identity, identity);
|
||||
assert.deepEqual(connection.withdrawals[0]?.publicationIds, [advertisement.publicationId]);
|
||||
assert.equal(session.state.status, "stopped");
|
||||
assert.ok(states.includes("connecting"));
|
||||
assert.ok(states.includes("withdrawing"));
|
||||
});
|
||||
|
||||
test("stopping during registration prevents a stale connection from publishing", async () => {
|
||||
const scheduler = new ManualScheduler();
|
||||
const transport = new FakeTransport(scheduler);
|
||||
transport.registerBarrier = new Deferred<void>();
|
||||
const session = new CultMeshProviderSession({
|
||||
identity,
|
||||
transport,
|
||||
scheduler,
|
||||
receiptStore: new CultMeshMemoryProviderReceiptStore(),
|
||||
publications: [advertisement],
|
||||
leaseDurationMs: 1_000,
|
||||
renewalLeadMs: 200,
|
||||
});
|
||||
|
||||
const starting = session.start();
|
||||
await settle();
|
||||
const connection = transport.connections[0]!;
|
||||
const stopping = session.stop();
|
||||
transport.registerBarrier.resolve();
|
||||
await Promise.all([starting, stopping]);
|
||||
|
||||
assert.equal(session.state.status, "stopped");
|
||||
assert.equal(connection.closed, true);
|
||||
assert.deepEqual(connection.publications, []);
|
||||
assert.deepEqual(connection.withdrawals, []);
|
||||
});
|
||||
|
||||
test("provider session reconnects and republishes desired state after transient failure", async () => {
|
||||
const scheduler = new ManualScheduler();
|
||||
const transport = new FakeTransport(scheduler);
|
||||
transport.failConnectCount = 2;
|
||||
const session = new CultMeshProviderSession({
|
||||
identity,
|
||||
transport,
|
||||
scheduler,
|
||||
receiptStore: new CultMeshMemoryProviderReceiptStore(),
|
||||
publications: [advertisement],
|
||||
leaseDurationMs: 1_000,
|
||||
renewalLeadMs: 200,
|
||||
reconnectBaseDelayMs: 100,
|
||||
reconnectMaxDelayMs: 400,
|
||||
});
|
||||
|
||||
await session.start();
|
||||
assert.equal(session.state.status, "reconnecting");
|
||||
assert.equal(session.state.reconnectAttempt, 1);
|
||||
|
||||
await scheduler.advance(100);
|
||||
assert.equal(session.state.status, "reconnecting");
|
||||
assert.equal(session.state.reconnectAttempt, 2);
|
||||
await scheduler.advance(200);
|
||||
assert.equal(session.state.status, "active");
|
||||
assert.equal(transport.connections.length, 1);
|
||||
assert.deepEqual(
|
||||
transport.connections[0]?.publications.map(value => value.publicationId),
|
||||
[advertisement.publicationId],
|
||||
);
|
||||
|
||||
transport.connections[0]!.failNextPublish = new Error("partition");
|
||||
await session.upsertPublication(surface);
|
||||
assert.equal(session.state.status, "reconnecting");
|
||||
assert.match(session.state.lastError ?? "", /partition/);
|
||||
assert.equal(transport.connections[0]?.closed, true);
|
||||
|
||||
await scheduler.advance(100);
|
||||
assert.equal(session.state.status, "active");
|
||||
assert.deepEqual(
|
||||
transport.connections[1]?.publications.map(value => value.publicationId),
|
||||
[advertisement.publicationId, surface.publicationId],
|
||||
);
|
||||
await session.stop();
|
||||
});
|
||||
|
||||
test("receipt persistence failure degrades the provider without inventing a transport outage", async () => {
|
||||
const scheduler = new ManualScheduler();
|
||||
const transport = new FakeTransport(scheduler);
|
||||
const session = new CultMeshProviderSession({
|
||||
identity,
|
||||
transport,
|
||||
scheduler,
|
||||
receiptStore: {
|
||||
get: async () => { throw new Error("receipt store offline"); },
|
||||
put: async () => undefined,
|
||||
listPending: async () => [],
|
||||
markPublished: async () => undefined,
|
||||
},
|
||||
leaseDurationMs: 1_000,
|
||||
renewalLeadMs: 200,
|
||||
commandHandlers: {
|
||||
"swarm.set_heat": async () => ({ state: "applied" }),
|
||||
},
|
||||
});
|
||||
await session.start();
|
||||
const connection = transport.connections[0]!;
|
||||
await connection.emit({
|
||||
commandId: "heat-store-failure",
|
||||
commandKind: "swarm.set_heat",
|
||||
providerId: identity.providerId,
|
||||
serviceInstanceId: identity.serviceInstanceId,
|
||||
payload: 0.5,
|
||||
});
|
||||
|
||||
assert.equal(session.state.status, "degraded");
|
||||
assert.match(session.state.lastError ?? "", /receipt store offline/);
|
||||
assert.equal(connection.closed, false);
|
||||
assert.equal(transport.connections.length, 1);
|
||||
await session.stop();
|
||||
});
|
||||
|
||||
test("provider session dispatches commands once and republishes typed receipts", async () => {
|
||||
const scheduler = new ManualScheduler();
|
||||
const transport = new FakeTransport(scheduler);
|
||||
let applyCount = 0;
|
||||
const session = new CultMeshProviderSession({
|
||||
identity,
|
||||
transport,
|
||||
scheduler,
|
||||
receiptStore: new CultMeshMemoryProviderReceiptStore(),
|
||||
leaseDurationMs: 1_000,
|
||||
renewalLeadMs: 200,
|
||||
commandHandlers: {
|
||||
"swarm.set_heat": async command => {
|
||||
applyCount++;
|
||||
await settle();
|
||||
return { state: "applied", result: { heat: command.payload } };
|
||||
},
|
||||
},
|
||||
});
|
||||
await session.start();
|
||||
const connection = transport.connections[0]!;
|
||||
const command: CultMeshProviderCommand = {
|
||||
commandId: "heat-1",
|
||||
commandKind: "swarm.set_heat",
|
||||
providerId: identity.providerId,
|
||||
serviceInstanceId: identity.serviceInstanceId,
|
||||
payload: 0.75,
|
||||
};
|
||||
|
||||
await Promise.all([connection.emit(command), connection.emit(command)]);
|
||||
await connection.emit(command);
|
||||
|
||||
assert.equal(applyCount, 1);
|
||||
assert.equal(connection.receipts.length, 2, "concurrent duplicates share one delivery; later retries republish");
|
||||
assert.ok(connection.receipts.every(receipt =>
|
||||
receipt.receiptId === "receipt:13:voidbot.swarm16:voidbot-worker-76:heat-1"));
|
||||
assert.ok(connection.receipts.every(receipt => receipt.state === "applied"));
|
||||
|
||||
await connection.emit({ ...command, commandId: "unknown-1", commandKind: "swarm.unknown" });
|
||||
assert.equal(connection.receipts.at(-1)?.state, "rejected");
|
||||
|
||||
await connection.emit({ ...command, commandId: "wrong-instance", serviceInstanceId: "somebody-else" });
|
||||
assert.equal(connection.receipts.length, 3, "commands for another service instance are ignored");
|
||||
await session.stop();
|
||||
});
|
||||
|
||||
test("persisted receipts survive a transport failure and leave the outbox only after replay", async () => {
|
||||
const scheduler = new ManualScheduler();
|
||||
const transport = new FakeTransport(scheduler);
|
||||
const receipts = new CultMeshMemoryProviderReceiptStore();
|
||||
let applyCount = 0;
|
||||
const session = new CultMeshProviderSession({
|
||||
identity,
|
||||
transport,
|
||||
scheduler,
|
||||
receiptStore: receipts,
|
||||
leaseDurationMs: 1_000,
|
||||
renewalLeadMs: 200,
|
||||
reconnectBaseDelayMs: 100,
|
||||
commandHandlers: {
|
||||
"swarm.set_heat": async () => {
|
||||
applyCount++;
|
||||
return { state: "applied" };
|
||||
},
|
||||
},
|
||||
});
|
||||
await session.start();
|
||||
transport.connections[0]!.failNextReceipt = new Error("receipt route lost");
|
||||
await transport.connections[0]!.emit({
|
||||
commandId: "durable-1",
|
||||
commandKind: "swarm.set_heat",
|
||||
providerId: identity.providerId,
|
||||
serviceInstanceId: identity.serviceInstanceId,
|
||||
payload: 0.25,
|
||||
});
|
||||
|
||||
assert.equal(session.state.status, "reconnecting");
|
||||
assert.equal((await receipts.listPending()).length, 1);
|
||||
await scheduler.advance(100);
|
||||
assert.equal(session.state.status, "active");
|
||||
assert.equal(transport.connections[1]!.receipts.length, 1);
|
||||
assert.equal((await receipts.listPending()).length, 0);
|
||||
assert.equal(applyCount, 1);
|
||||
|
||||
transport.connections[1]!.failNextPublish = new Error("state route lost");
|
||||
await session.upsertPublication(surface);
|
||||
await scheduler.advance(100);
|
||||
assert.equal(transport.connections[2]!.receipts.length, 0, "marked receipts are not replayed again");
|
||||
await session.stop();
|
||||
});
|
||||
|
||||
test("initial replay and live publication changes share one ordered lane", async () => {
|
||||
const scheduler = new ManualScheduler();
|
||||
const transport = new FakeTransport(scheduler);
|
||||
transport.publishBarrier = new Deferred<void>();
|
||||
const session = new CultMeshProviderSession({
|
||||
identity,
|
||||
transport,
|
||||
scheduler,
|
||||
receiptStore: new CultMeshMemoryProviderReceiptStore(),
|
||||
publications: [surface],
|
||||
leaseDurationMs: 1_000,
|
||||
renewalLeadMs: 200,
|
||||
});
|
||||
|
||||
const starting = session.start();
|
||||
await settle();
|
||||
const connection = transport.connections[0]!;
|
||||
const updated = { ...surface, value: { version: 2 } };
|
||||
const updating = session.upsertPublication(updated);
|
||||
const removing = session.removePublication(surface.publicationId);
|
||||
transport.publishBarrier.resolve();
|
||||
await Promise.all([starting, updating, removing]);
|
||||
|
||||
assert.deepEqual(connection.publications, [surface, updated]);
|
||||
assert.deepEqual(connection.withdrawnPublications, [surface.publicationId]);
|
||||
assert.deepEqual(session.publications, []);
|
||||
await session.stop();
|
||||
});
|
||||
|
||||
test("state observers cannot become lifecycle authorities", async () => {
|
||||
const scheduler = new ManualScheduler();
|
||||
const transport = new FakeTransport(scheduler);
|
||||
const observerErrors: unknown[] = [];
|
||||
const observed: string[] = [];
|
||||
const session = new CultMeshProviderSession({
|
||||
identity,
|
||||
transport,
|
||||
scheduler,
|
||||
receiptStore: new CultMeshMemoryProviderReceiptStore(),
|
||||
leaseDurationMs: 1_000,
|
||||
renewalLeadMs: 200,
|
||||
onObserverError: error => observerErrors.push(error),
|
||||
});
|
||||
session.watchState(() => { throw new Error("display broken"); });
|
||||
session.watchState(state => observed.push(state.status));
|
||||
|
||||
await session.start();
|
||||
await session.stop();
|
||||
|
||||
assert.equal(session.state.status, "stopped");
|
||||
assert.ok(observerErrors.length >= 3);
|
||||
assert.deepEqual(observed, ["stopped", "connecting", "active", "withdrawing", "stopped"]);
|
||||
});
|
||||
|
||||
test("shutdown drains command persistence and always closes after cleanup failures", async () => {
|
||||
const scheduler = new ManualScheduler();
|
||||
const transport = new FakeTransport(scheduler);
|
||||
const handlerBarrier = new Deferred<void>();
|
||||
const session = new CultMeshProviderSession({
|
||||
identity,
|
||||
transport,
|
||||
scheduler,
|
||||
receiptStore: new CultMeshMemoryProviderReceiptStore(),
|
||||
leaseDurationMs: 1_000,
|
||||
renewalLeadMs: 200,
|
||||
commandHandlers: {
|
||||
"swarm.set_heat": async () => {
|
||||
await handlerBarrier.promise;
|
||||
return { state: "applied" };
|
||||
},
|
||||
},
|
||||
});
|
||||
await session.start();
|
||||
const connection = transport.connections[0]!;
|
||||
const command = connection.emit({
|
||||
commandId: "stop-drain-1",
|
||||
commandKind: "swarm.set_heat",
|
||||
providerId: identity.providerId,
|
||||
serviceInstanceId: identity.serviceInstanceId,
|
||||
payload: 0.5,
|
||||
});
|
||||
connection.unsubscribeError = new Error("unsubscribe failed");
|
||||
connection.withdrawError = new Error("withdraw failed");
|
||||
connection.closeError = new Error("close failed");
|
||||
let stopped = false;
|
||||
const stopping = session.stop().finally(() => { stopped = true; });
|
||||
await settle();
|
||||
assert.equal(stopped, false, "shutdown waits for the authoritative command transaction");
|
||||
handlerBarrier.resolve();
|
||||
await command;
|
||||
await assert.rejects(stopping, AggregateError);
|
||||
|
||||
assert.equal(connection.closeAttempts, 1);
|
||||
assert.equal(session.state.status, "stopped");
|
||||
});
|
||||
|
||||
test("concurrent reuse of a command id with another kind cannot join the first transaction", async () => {
|
||||
const scheduler = new ManualScheduler();
|
||||
const transport = new FakeTransport(scheduler);
|
||||
const barrier = new Deferred<void>();
|
||||
let heatCount = 0;
|
||||
let otherCount = 0;
|
||||
const session = new CultMeshProviderSession({
|
||||
identity,
|
||||
transport,
|
||||
scheduler,
|
||||
receiptStore: new CultMeshMemoryProviderReceiptStore(),
|
||||
leaseDurationMs: 1_000,
|
||||
renewalLeadMs: 200,
|
||||
commandHandlers: {
|
||||
"swarm.set_heat": async () => {
|
||||
heatCount++;
|
||||
await barrier.promise;
|
||||
return { state: "applied" };
|
||||
},
|
||||
"swarm.other": async () => {
|
||||
otherCount++;
|
||||
return { state: "applied" };
|
||||
},
|
||||
},
|
||||
});
|
||||
await session.start();
|
||||
const connection = transport.connections[0]!;
|
||||
const first = connection.emit({
|
||||
commandId: "collision-1",
|
||||
commandKind: "swarm.set_heat",
|
||||
providerId: identity.providerId,
|
||||
serviceInstanceId: identity.serviceInstanceId,
|
||||
payload: 0.5,
|
||||
});
|
||||
await settle();
|
||||
await connection.emit({
|
||||
commandId: "collision-1",
|
||||
commandKind: "swarm.other",
|
||||
providerId: identity.providerId,
|
||||
serviceInstanceId: identity.serviceInstanceId,
|
||||
payload: null,
|
||||
});
|
||||
assert.equal(session.state.status, "degraded");
|
||||
barrier.resolve();
|
||||
await first;
|
||||
assert.equal(heatCount, 1);
|
||||
assert.equal(otherCount, 0);
|
||||
await session.stop();
|
||||
});
|
||||
|
||||
test("registration failure closes the connection before retry", async () => {
|
||||
const scheduler = new ManualScheduler();
|
||||
const transport = new FakeTransport(scheduler);
|
||||
transport.registerError = new Error("registration rejected");
|
||||
const session = new CultMeshProviderSession({
|
||||
identity,
|
||||
transport,
|
||||
scheduler,
|
||||
receiptStore: new CultMeshMemoryProviderReceiptStore(),
|
||||
leaseDurationMs: 1_000,
|
||||
renewalLeadMs: 200,
|
||||
reconnectBaseDelayMs: 100,
|
||||
});
|
||||
await session.start();
|
||||
assert.equal(transport.connections[0]!.closed, true);
|
||||
assert.equal(session.state.status, "reconnecting");
|
||||
await session.stop();
|
||||
});
|
||||
|
||||
test("an expired replay lease never carries the next pending receipt", async () => {
|
||||
const scheduler = new ManualScheduler();
|
||||
const transport = new FakeTransport(scheduler);
|
||||
const receipts = new CultMeshMemoryProviderReceiptStore();
|
||||
const pending = (commandId: string): CultMeshProviderCommandReceipt => ({
|
||||
receiptId: `receipt:13:voidbot.swarm16:voidbot-worker-7${commandId.length}:${commandId}`,
|
||||
commandId,
|
||||
commandKind: "swarm.set_heat",
|
||||
providerId: identity.providerId,
|
||||
serviceInstanceId: identity.serviceInstanceId,
|
||||
state: "applied",
|
||||
completedAt: scheduler.now(),
|
||||
});
|
||||
await receipts.put(pending("one"));
|
||||
await receipts.put(pending("two"));
|
||||
const session = new CultMeshProviderSession({
|
||||
identity,
|
||||
transport,
|
||||
scheduler,
|
||||
receiptStore: receipts,
|
||||
leaseDurationMs: 1_000,
|
||||
renewalLeadMs: 200,
|
||||
reconnectBaseDelayMs: 100,
|
||||
});
|
||||
transport.onFirstConnection = connection => {
|
||||
connection.afterPublishReceipt = async () => scheduler.advance(1_001);
|
||||
};
|
||||
|
||||
await session.start();
|
||||
assert.equal(session.state.status, "reconnecting");
|
||||
assert.deepEqual(
|
||||
transport.connections[0]!.receipts.map(value => value.commandId),
|
||||
["one"],
|
||||
"the expired lease is rejected before the next receipt",
|
||||
);
|
||||
assert.deepEqual((await receipts.listPending()).map(value => value.commandId), ["two"]);
|
||||
await scheduler.advance(100);
|
||||
assert.equal(session.state.status, "active");
|
||||
assert.deepEqual(transport.connections[1]!.receipts.map(value => value.commandId), ["two"]);
|
||||
assert.equal((await receipts.listPending()).length, 0);
|
||||
await session.stop();
|
||||
});
|
||||
|
||||
test("overlapping shutdown callers share cleanup and restart waits for its boundary", async () => {
|
||||
const scheduler = new ManualScheduler();
|
||||
const transport = new FakeTransport(scheduler);
|
||||
const session = new CultMeshProviderSession({
|
||||
identity,
|
||||
transport,
|
||||
scheduler,
|
||||
receiptStore: new CultMeshMemoryProviderReceiptStore(),
|
||||
leaseDurationMs: 1_000,
|
||||
renewalLeadMs: 200,
|
||||
});
|
||||
await session.start();
|
||||
const connection = transport.connections[0]!;
|
||||
connection.withdrawBarrier = new Deferred<void>();
|
||||
|
||||
const firstStop = session.stop();
|
||||
const secondStop = session.stop();
|
||||
assert.equal(firstStop, secondStop, "concurrent callers own one shutdown operation");
|
||||
await assert.rejects(session.start(), /shutdown is in progress/);
|
||||
await settle();
|
||||
assert.equal(connection.closed, false);
|
||||
connection.withdrawBarrier.resolve();
|
||||
await firstStop;
|
||||
assert.equal(session.state.status, "stopped");
|
||||
|
||||
await session.start();
|
||||
assert.equal(session.state.status, "active");
|
||||
assert.equal(transport.connections.length, 2);
|
||||
await session.stop();
|
||||
});
|
||||
|
||||
test("loss cleanup failures cannot suppress reconnect scheduling", async () => {
|
||||
const scheduler = new ManualScheduler();
|
||||
const transport = new FakeTransport(scheduler);
|
||||
const session = new CultMeshProviderSession({
|
||||
identity,
|
||||
transport,
|
||||
scheduler,
|
||||
receiptStore: new CultMeshMemoryProviderReceiptStore(),
|
||||
leaseDurationMs: 1_000,
|
||||
renewalLeadMs: 200,
|
||||
reconnectBaseDelayMs: 100,
|
||||
});
|
||||
await session.start();
|
||||
const connection = transport.connections[0]!;
|
||||
connection.unsubscribeError = new Error("watch cleanup failed");
|
||||
connection.closeError = new Error("route close failed");
|
||||
connection.failNextPublish = new Error("route lost");
|
||||
|
||||
await session.upsertPublication(surface);
|
||||
assert.equal(session.state.status, "reconnecting");
|
||||
assert.match(session.state.lastError ?? "", /watch cleanup failed/);
|
||||
assert.match(session.state.lastError ?? "", /route close failed/);
|
||||
await scheduler.advance(100);
|
||||
assert.equal(session.state.status, "active");
|
||||
await session.stop();
|
||||
});
|
||||
|
||||
test("stop during receipt replay cannot publish active state or leave a renewal timer", async () => {
|
||||
const scheduler = new ManualScheduler();
|
||||
const transport = new FakeTransport(scheduler);
|
||||
const listBarrier = new Deferred<void>();
|
||||
const states: string[] = [];
|
||||
const session = new CultMeshProviderSession({
|
||||
identity,
|
||||
transport,
|
||||
scheduler,
|
||||
receiptStore: {
|
||||
get: async () => undefined,
|
||||
put: async () => undefined,
|
||||
listPending: async () => {
|
||||
await listBarrier.promise;
|
||||
return [];
|
||||
},
|
||||
markPublished: async () => undefined,
|
||||
},
|
||||
leaseDurationMs: 1_000,
|
||||
renewalLeadMs: 200,
|
||||
});
|
||||
session.watchState(state => states.push(state.status));
|
||||
const starting = session.start();
|
||||
await settle();
|
||||
const stopping = session.stop();
|
||||
await settle();
|
||||
const withdrawingIndex = states.lastIndexOf("withdrawing");
|
||||
listBarrier.resolve();
|
||||
await Promise.all([starting, stopping]);
|
||||
|
||||
assert.equal(session.state.status, "stopped");
|
||||
assert.equal(states.slice(withdrawingIndex + 1).includes("active"), false);
|
||||
assert.equal(scheduler.pendingTaskCount, 0);
|
||||
});
|
||||
|
||||
class ManualScheduler implements CultMeshProviderScheduler {
|
||||
#nowMs = Date.parse("2026-07-14T00:00:00.000Z");
|
||||
readonly #tasks: Array<{
|
||||
dueAt: number;
|
||||
cancelled: boolean;
|
||||
action: () => void | Promise<void>;
|
||||
}> = [];
|
||||
|
||||
public now(): Date {
|
||||
return new Date(this.#nowMs);
|
||||
}
|
||||
|
||||
public get pendingTaskCount(): number {
|
||||
return this.#tasks.filter(value => !value.cancelled).length;
|
||||
}
|
||||
|
||||
public schedule(delayMs: number, action: () => void | Promise<void>): CultMeshProviderScheduledTask {
|
||||
const task = { dueAt: this.#nowMs + delayMs, cancelled: false, action };
|
||||
this.#tasks.push(task);
|
||||
return { cancel: () => { task.cancelled = true; } };
|
||||
}
|
||||
|
||||
public async advance(delayMs: number): Promise<void> {
|
||||
this.#nowMs += delayMs;
|
||||
while (true) {
|
||||
const task = this.#tasks
|
||||
.filter(value => !value.cancelled && value.dueAt <= this.#nowMs)
|
||||
.sort((left, right) => left.dueAt - right.dueAt)[0];
|
||||
if (!task) return;
|
||||
task.cancelled = true;
|
||||
await task.action();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
class FakeTransport implements CultMeshProviderTransport {
|
||||
public failConnectCount = 0;
|
||||
public registerBarrier?: Deferred<void>;
|
||||
public registerError?: Error;
|
||||
public publishBarrier?: Deferred<void>;
|
||||
public onFirstConnection?: (connection: FakeConnection) => void;
|
||||
public readonly connections: FakeConnection[] = [];
|
||||
|
||||
public constructor(private readonly scheduler: CultMeshProviderScheduler) {}
|
||||
|
||||
public async connect(
|
||||
connectedIdentity: CultMeshProviderIdentity,
|
||||
signal: AbortSignal,
|
||||
): Promise<CultMeshProviderConnection> {
|
||||
assert.deepEqual(connectedIdentity, identity);
|
||||
assert.equal(signal.aborted, false);
|
||||
if (this.failConnectCount-- > 0) throw new Error("bootstrap unavailable");
|
||||
const connection = new FakeConnection(this.scheduler);
|
||||
connection.registerBarrier = this.registerBarrier;
|
||||
connection.registerError = this.registerError;
|
||||
connection.publishBarrier = this.publishBarrier;
|
||||
this.connections.push(connection);
|
||||
if (this.connections.length === 1) this.onFirstConnection?.(connection);
|
||||
return connection;
|
||||
}
|
||||
}
|
||||
|
||||
class FakeConnection implements CultMeshProviderConnection {
|
||||
public readonly registrations: CultMeshProviderRegistration[] = [];
|
||||
public readonly renewals: CultMeshProviderLease[] = [];
|
||||
public readonly publications: CultMeshProviderPublication[] = [];
|
||||
public readonly withdrawnPublications: string[] = [];
|
||||
public readonly receipts: CultMeshProviderCommandReceipt[] = [];
|
||||
public readonly withdrawals: CultMeshProviderWithdrawal[] = [];
|
||||
public failNextPublish?: Error;
|
||||
public failNextReceipt?: Error;
|
||||
public registerError?: Error;
|
||||
public registerBarrier?: Deferred<void>;
|
||||
public publishBarrier?: Deferred<void>;
|
||||
public unsubscribeError?: Error;
|
||||
public withdrawError?: Error;
|
||||
public closeError?: Error;
|
||||
public closeAttempts = 0;
|
||||
public afterPublishReceipt?: () => void | Promise<void>;
|
||||
public withdrawBarrier?: Deferred<void>;
|
||||
public closed = false;
|
||||
#listener?: CultMeshProviderCommandListener;
|
||||
#leaseSequence = 0;
|
||||
|
||||
public constructor(private readonly scheduler: CultMeshProviderScheduler) {}
|
||||
|
||||
public async register(
|
||||
registration: CultMeshProviderRegistration,
|
||||
signal: AbortSignal,
|
||||
): Promise<CultMeshProviderLease> {
|
||||
this.registrations.push(registration);
|
||||
if (this.registerBarrier) {
|
||||
await Promise.race([
|
||||
this.registerBarrier.promise,
|
||||
new Promise<never>((_resolve, reject) => {
|
||||
signal.addEventListener("abort", () => reject(new Error("registration aborted")), { once: true });
|
||||
}),
|
||||
]);
|
||||
}
|
||||
if (this.registerError) throw this.registerError;
|
||||
return this.lease(registration.requestedLeaseDurationMs);
|
||||
}
|
||||
|
||||
public async renew(lease: CultMeshProviderLease): Promise<CultMeshProviderLease> {
|
||||
this.renewals.push(lease);
|
||||
return this.lease(1_000);
|
||||
}
|
||||
|
||||
public async publish(publication: CultMeshProviderPublication): Promise<void> {
|
||||
if (this.failNextPublish) {
|
||||
const error = this.failNextPublish;
|
||||
this.failNextPublish = undefined;
|
||||
throw error;
|
||||
}
|
||||
const barrier = this.publishBarrier;
|
||||
this.publishBarrier = undefined;
|
||||
await barrier?.promise;
|
||||
this.publications.push(publication);
|
||||
}
|
||||
|
||||
public async withdrawPublication(publicationId: string): Promise<void> {
|
||||
this.withdrawnPublications.push(publicationId);
|
||||
}
|
||||
|
||||
public watchCommands(listener: CultMeshProviderCommandListener): CultMeshProviderUnsubscribe {
|
||||
this.#listener = listener;
|
||||
return () => {
|
||||
this.#listener = undefined;
|
||||
if (this.unsubscribeError) throw this.unsubscribeError;
|
||||
};
|
||||
}
|
||||
|
||||
public async publishReceipt(receipt: CultMeshProviderCommandReceipt): Promise<void> {
|
||||
if (this.failNextReceipt) {
|
||||
const error = this.failNextReceipt;
|
||||
this.failNextReceipt = undefined;
|
||||
throw error;
|
||||
}
|
||||
this.receipts.push(receipt);
|
||||
const afterPublish = this.afterPublishReceipt;
|
||||
this.afterPublishReceipt = undefined;
|
||||
await afterPublish?.();
|
||||
}
|
||||
|
||||
public async withdraw(withdrawal: CultMeshProviderWithdrawal): Promise<void> {
|
||||
this.withdrawals.push(withdrawal);
|
||||
await this.withdrawBarrier?.promise;
|
||||
if (this.withdrawError) throw this.withdrawError;
|
||||
}
|
||||
|
||||
public close(): void {
|
||||
this.closeAttempts++;
|
||||
this.closed = true;
|
||||
if (this.closeError) throw this.closeError;
|
||||
}
|
||||
|
||||
public async emit(command: CultMeshProviderCommand): Promise<void> {
|
||||
await this.#listener?.(command);
|
||||
}
|
||||
|
||||
private lease(durationMs: number): CultMeshProviderLease {
|
||||
this.#leaseSequence++;
|
||||
return {
|
||||
leaseId: `lease-${this.#leaseSequence}`,
|
||||
expiresAt: new Date(this.scheduler.now().getTime() + durationMs),
|
||||
};
|
||||
}
|
||||
}
|
||||
|
||||
async function settle(): Promise<void> {
|
||||
await new Promise<void>(resolve => setImmediate(resolve));
|
||||
}
|
||||
|
||||
class Deferred<T> {
|
||||
public readonly promise: Promise<T>;
|
||||
#resolve!: (value: T | PromiseLike<T>) => void;
|
||||
|
||||
public constructor() {
|
||||
this.promise = new Promise<T>(resolve => { this.#resolve = resolve; });
|
||||
}
|
||||
|
||||
public resolve(value?: T): void {
|
||||
this.#resolve(value as T);
|
||||
}
|
||||
}
|
||||
|
|
@ -64,6 +64,7 @@ const mesh = require("cultmesh-ts");
|
|||
assert.equal(typeof cache.defineDocumentType, "function");
|
||||
assert.equal(typeof net.CultNetPeer, "function");
|
||||
assert.equal(typeof mesh.CultMesh, "function");
|
||||
assert.equal(typeof mesh.CultMeshProviderSession, "function");
|
||||
assert.deepEqual(mesh.CultMesh.vec2(3, 4), { x: 3, y: 4 });
|
||||
`);
|
||||
execFileSync(process.execPath, ["runtime-smoke.cjs"], { cwd: consumerRoot, stdio: "inherit" });
|
||||
|
|
@ -71,14 +72,31 @@ assert.deepEqual(mesh.CultMesh.vec2(3, 4), { x: 3, y: 4 });
|
|||
writeFileSync(join(consumerRoot, "types-smoke.ts"), `
|
||||
import { defineDocumentType } from "cultcache-ts";
|
||||
import { CultNetPeer } from "cultnet-ts";
|
||||
import { CultMesh, type CultMeshVec2 } from "cultmesh-ts";
|
||||
import {
|
||||
CultMesh,
|
||||
CultMeshProviderSession,
|
||||
type CultMeshProviderIdentity,
|
||||
type CultMeshProviderTransport,
|
||||
type CultMeshVec2,
|
||||
} from "cultmesh-ts";
|
||||
|
||||
const point: CultMeshVec2 = CultMesh.vec2(3, 4);
|
||||
const defineDocument: typeof defineDocumentType = defineDocumentType;
|
||||
const peerType: typeof CultNetPeer = CultNetPeer;
|
||||
const providerSessionType: typeof CultMeshProviderSession = CultMeshProviderSession;
|
||||
const providerIdentity: CultMeshProviderIdentity = {
|
||||
providerId: "smoke.provider",
|
||||
serviceInstanceId: "smoke.instance",
|
||||
endpointId: "odin:smoke.instance",
|
||||
verseId: "smoke",
|
||||
};
|
||||
let providerTransport: CultMeshProviderTransport | undefined;
|
||||
void point;
|
||||
void defineDocument;
|
||||
void peerType;
|
||||
void providerSessionType;
|
||||
void providerIdentity;
|
||||
void providerTransport;
|
||||
`);
|
||||
const tsc = join(repoRoot, "node_modules", "typescript", "bin", "tsc");
|
||||
assert.ok(readFileSync(tsc, "utf8").length > 0, "workspace TypeScript compiler is unavailable");
|
||||
|
|
|
|||
|
|
@ -0,0 +1,122 @@
|
|||
# 4. Keep A TypeScript Provider Alive
|
||||
|
||||
A provider process should declare desired typed publications and domain command
|
||||
handlers once. It should not rebuild discovery registration, lease renewal,
|
||||
reconnect, republish, receipt delivery, and withdrawal loops in a renderer or
|
||||
scheduled export script.
|
||||
|
||||
`CultMeshProviderSession` owns that lifecycle. The application still owns its
|
||||
state and command transaction. A transport connector owns the physical CultNet
|
||||
route. Eve advertisements and surfaces are ordinary typed publications; the
|
||||
session does not import Eve or learn provider gameplay.
|
||||
|
||||
```ts
|
||||
import {
|
||||
CultMeshProviderSession,
|
||||
type CultMeshProviderReceiptStore,
|
||||
type CultMeshProviderTransport,
|
||||
} from "cultmesh-ts";
|
||||
|
||||
export async function runProvider(
|
||||
transport: CultMeshProviderTransport,
|
||||
receipts: CultMeshProviderReceiptStore,
|
||||
) {
|
||||
const provider = new CultMeshProviderSession({
|
||||
identity: {
|
||||
providerId: "voidbot.swarm",
|
||||
serviceInstanceId: "voidbot-worker-7",
|
||||
endpointId: "odin:voidbot-worker-7",
|
||||
verseId: "voidbot.local",
|
||||
},
|
||||
transport,
|
||||
receiptStore: receipts,
|
||||
publications: [
|
||||
{
|
||||
publicationId: "voidbot.swarm.advertisement",
|
||||
documentType: "gamecult.eve.provider_advertisement",
|
||||
schemaId: "gamecult.eve.provider_advertisement.v1",
|
||||
recordKey: "voidbot.swarm",
|
||||
value: providerAdvertisement,
|
||||
},
|
||||
{
|
||||
publicationId: "voidbot.swarm.surface",
|
||||
documentType: "gamecult.eve.surface_state",
|
||||
schemaId: "gamecult.eve.surface_state.v1",
|
||||
recordKey: "voidbot.swarm",
|
||||
value: currentSurface,
|
||||
},
|
||||
],
|
||||
commandHandlers: {
|
||||
"swarm.set_heat": async command =>
|
||||
applyHeatCommandTransaction(command.commandId, command.payload),
|
||||
},
|
||||
});
|
||||
|
||||
await provider.start();
|
||||
|
||||
await provider.upsertPublication({
|
||||
publicationId: "voidbot.swarm.surface",
|
||||
documentType: "gamecult.eve.surface_state",
|
||||
schemaId: "gamecult.eve.surface_state.v1",
|
||||
recordKey: "voidbot.swarm",
|
||||
value: nextSurface,
|
||||
});
|
||||
|
||||
process.once("SIGTERM", () => {
|
||||
provider.stop().catch(error => {
|
||||
console.error("CultMesh provider shutdown completed with cleanup failures.", error);
|
||||
process.exitCode = 1;
|
||||
});
|
||||
});
|
||||
}
|
||||
```
|
||||
|
||||
The four identities are deliberately distinct:
|
||||
|
||||
- `providerId` names the product capability clients select;
|
||||
- `serviceInstanceId` names the running owner incarnation;
|
||||
- `endpointId` names the stable discovery/session route.
|
||||
- `verseId` names the discovery and state-sharing domain.
|
||||
|
||||
Do not replace any of them with a process id, socket address, body producer id,
|
||||
or renderer id.
|
||||
|
||||
## Receipt Rule
|
||||
|
||||
The receipt store is required and must be durable in a daemon. The in-memory
|
||||
store exported by the package is for tests and disposable tools only. The
|
||||
domain command handler must apply its mutation idempotently using `commandId`;
|
||||
the provider session deduplicates delivery and republishes the stored typed
|
||||
receipt, but it cannot make an unrelated database mutation atomic by wishing
|
||||
at it harder. `commandId` is the idempotency key: later delivery of the same
|
||||
provider, service-instance, and command kind returns the original receipt even
|
||||
if a sender incorrectly changes the payload. Reusing the ID for another kind
|
||||
or identity is rejected as a conflict.
|
||||
|
||||
The store is also the receipt outbox. `put` commits the receipt before any wire
|
||||
publication. `listPending` supplies receipts after startup or reconnect, and
|
||||
`markPublished` removes one only after the active connection accepts it. Scope
|
||||
the store to the provider and service-instance identity. If the store is
|
||||
unavailable, the session enters `degraded` and stops accepting commands until a
|
||||
renewal can drain the outbox again; a store failure is not a transport failure.
|
||||
|
||||
## Verification Checkpoint
|
||||
|
||||
Use an injected scheduler and transport fault double to prove the actual
|
||||
lifecycle:
|
||||
|
||||
1. initial registration publishes every desired document in stable order;
|
||||
2. lease renewal occurs before expiry;
|
||||
3. a failed route produces `reconnecting`, then republishes current desired
|
||||
state on the replacement connection;
|
||||
4. duplicate command delivery runs one handler transaction and republishes the
|
||||
same receipt;
|
||||
5. failed receipt transport reconnects and replays the durable outbox exactly
|
||||
until `markPublished` succeeds;
|
||||
6. shutdown drains command persistence, withdraws the lease and remaining
|
||||
publications, and closes even when cleanup reports failures;
|
||||
7. stopping during registration cannot let the stale connection publish;
|
||||
8. observers and conflicting command IDs cannot become lifecycle authorities.
|
||||
|
||||
The package suite contains this timeline in
|
||||
`packages/cultmesh-ts/test/provider-session.test.ts`.
|
||||
|
|
@ -9,6 +9,7 @@ Follow the chapters in order:
|
|||
1. [Persist typed application state](01-persist-typed-state.md)
|
||||
2. [Connect a client by stable identity](02-connect-by-identity.md)
|
||||
3. [Publish one Eve surface to multiple runtimes](03-publish-an-eve-surface.md)
|
||||
4. [Keep a TypeScript provider alive](04-keep-a-typescript-provider-alive.md)
|
||||
|
||||
The intended application shape is small:
|
||||
|
||||
|
|
@ -33,4 +34,3 @@ native projection. CultMesh owns discovery and connection continuity.
|
|||
pull request.
|
||||
- A setup step that depends on repository internals is an API defect, not
|
||||
acceptable tutorial folklore.
|
||||
|
||||
|
|
|
|||
|
|
@ -496,6 +496,54 @@ after bounded replay. The Aetheria/EveUnity witness proves provider
|
|||
discovery, live SoA state, commands, receipts, and assets through the migrated
|
||||
session path.
|
||||
|
||||
### Phase 2 TypeScript provider-lifecycle slice
|
||||
|
||||
**Owner:** `CultMeshProviderSession` owns one TypeScript provider's discovery
|
||||
registration, lease renewal, physical reconnect, desired typed publication
|
||||
replay, command dispatch, receipt publication, and explicit withdrawal. It is
|
||||
keyed by provider, service-instance, endpoint, and Verse identity; those
|
||||
identities are never collapsed.
|
||||
|
||||
**Inputs:** an injected `CultMeshProviderTransport`, stable provider identity,
|
||||
lease/reconnect policy, desired typed publications, domain command handlers, a
|
||||
required durable receipt store, and an injected scheduler.
|
||||
|
||||
**Outputs:** watchable connecting/active/reconnecting/withdrawing/stopped
|
||||
state, current lease evidence, ordered document publication, a durable typed
|
||||
receipt outbox, and withdrawal of the lease plus remaining publications.
|
||||
|
||||
**Derived state:** reconnect attempt, backoff, current physical connection, and
|
||||
the in-flight duplicate-command set. Pending receipt delivery is derived from
|
||||
the durable receipt store, not volatile session memory. Provider state, Eve surface semantics,
|
||||
command effects, endpoint discovery evidence, and body-producer identity are
|
||||
not derived by the session.
|
||||
|
||||
**Forbidden writers:** renderers, HTML exporters, and scheduled projection
|
||||
scripts may not register providers, renew leases, reconnect, publish receipts,
|
||||
or withdraw provider state. A transport connection cannot mutate provider
|
||||
domain state or treat successful delivery as authorization.
|
||||
|
||||
**Shared paths:** initial connect and reconnect both register, subscribe, and
|
||||
replay the same desired publication set. Live publication updates use one
|
||||
serialized lane, so an older update cannot arrive after a newer one. Duplicate
|
||||
command deliveries share one in-flight transaction and later deliveries reuse
|
||||
the durable receipt. A receipt is stored before publication and marked only
|
||||
after acceptance; reconnect drains pending receipts before command intake.
|
||||
|
||||
**Cut line:** TypeScript one-shot RUDP publication helpers remain low-level
|
||||
compatibility surfaces, not provider lifecycle owners. VoidBot's swarm renderer
|
||||
is the first identified obsolete owner; its provider catalog mutation,
|
||||
publication, reconnect, and receipt paths must be deleted when it adopts this
|
||||
session. That consumer migration is still open and must not run both owners.
|
||||
|
||||
**Verification layer:** deterministic TypeScript tests prove initial
|
||||
registration, renewal, stable publication order, reconnect and replay,
|
||||
duplicate receipt reuse, explicit withdrawal, and the negative stop-during-
|
||||
registration timeline, durable receipt replay, observer isolation, conflicting
|
||||
command IDs, and exception-safe shutdown. The empty-consumer tarball smoke proves the session is
|
||||
present in the installable `cultmesh-ts` package. A live Odin transport adapter
|
||||
and the VoidBot migration remain required for the production proof.
|
||||
|
||||
## Phase 3: Authority as a Mandatory Boundary
|
||||
|
||||
### Cut first
|
||||
|
|
|
|||
Loading…
Reference in a new issue