diff --git a/packages/cultmesh-ts/README.md b/packages/cultmesh-ts/README.md index 279e271e..2409898e 100644 --- a/packages/cultmesh-ts/README.md +++ b/packages/cultmesh-ts/README.md @@ -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 diff --git a/packages/cultmesh-ts/src/index.ts b/packages/cultmesh-ts/src/index.ts index c61dfee1..eca3222d 100644 --- a/packages/cultmesh-ts/src/index.ts +++ b/packages/cultmesh-ts/src/index.ts @@ -32,6 +32,8 @@ import { type CultNetWireContract, } from "cultnet-ts"; +export * from "./provider-session"; + export interface CultMeshVec2 { readonly x: number; readonly y: number; diff --git a/packages/cultmesh-ts/src/provider-session.ts b/packages/cultmesh-ts/src/provider-session.ts new file mode 100644 index 00000000..f63d222c --- /dev/null +++ b/packages/cultmesh-ts/src/provider-session.ts @@ -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 { + readonly publicationId: string; + readonly documentType: string; + readonly schemaId: string; + readonly recordKey: string; + readonly value: TValue; +} + +export interface CultMeshProviderCommand { + readonly commandId: string; + readonly commandKind: string; + readonly providerId: string; + readonly serviceInstanceId: string; + readonly payload: TPayload; +} + +export type CultMeshProviderReceiptState = "applied" | "rejected" | "failed"; + +export interface CultMeshProviderCommandReceipt { + 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 { + readonly state: CultMeshProviderReceiptState; + readonly result?: TResult; + readonly error?: string; +} + +export interface CultMeshProviderCommandContext { + readonly identity: CultMeshProviderIdentity; +} + +export type CultMeshProviderCommandHandler = ( + command: CultMeshProviderCommand, + context: CultMeshProviderCommandContext, +) => Promise; + +export interface CultMeshProviderReceiptStore { + get(commandId: string): Promise; + put(receipt: CultMeshProviderCommandReceipt): Promise; + listPending(): Promise; + markPublished(receiptId: string): Promise; +} + +export class CultMeshMemoryProviderReceiptStore implements CultMeshProviderReceiptStore { + readonly #receipts = new Map(); + readonly #pendingReceiptIds = new Set(); + + public async get(commandId: string): Promise { + return this.#receipts.get(commandId); + } + + public async put(receipt: CultMeshProviderCommandReceipt): Promise { + this.#receipts.set(receipt.commandId, receipt); + this.#pendingReceiptIds.add(receipt.receiptId); + } + + public async listPending(): Promise { + 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 { + 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; + +export type CultMeshProviderUnsubscribe = () => void | Promise; + +export interface CultMeshProviderConnection { + register( + registration: CultMeshProviderRegistration, + signal: AbortSignal, + ): Promise; + renew(lease: CultMeshProviderLease): Promise; + publish( + publication: CultMeshProviderPublication, + lease: CultMeshProviderLease, + ): Promise; + withdrawPublication(publicationId: string, lease: CultMeshProviderLease): Promise; + watchCommands(listener: CultMeshProviderCommandListener): CultMeshProviderUnsubscribe; + publishReceipt( + receipt: CultMeshProviderCommandReceipt, + lease: CultMeshProviderLease, + ): Promise; + withdraw(withdrawal: CultMeshProviderWithdrawal): Promise; + close(): void | Promise; +} + +export interface CultMeshProviderTransport { + connect( + identity: CultMeshProviderIdentity, + signal: AbortSignal, + ): Promise; +} + +export interface CultMeshProviderScheduledTask { + cancel(): void; +} + +export interface CultMeshProviderScheduler { + now(): Date; + schedule(delayMs: number, action: () => void | Promise): CultMeshProviderScheduledTask; +} + +export class CultMeshSystemProviderScheduler implements CultMeshProviderScheduler { + public now(): Date { + return new Date(); + } + + public schedule(delayMs: number, action: () => void | Promise): 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; + readonly commandHandlers?: Readonly>; + 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(); + readonly #handlers = new Map(); + readonly #subscribers = new Set<(state: CultMeshProviderSessionState) => void>(); + readonly #inFlightCommands = new Map; + }>(); + #state: CultMeshProviderSessionState; + #connection?: CultMeshProviderConnection; + #lease?: CultMeshProviderLease; + #unsubscribeCommands?: CultMeshProviderUnsubscribe; + #scheduled?: CultMeshProviderScheduledTask; + #abort?: AbortController; + #running = false; + #generation = 0; + #reconnectAttempt = 0; + #publicationWork: Promise = Promise.resolve(); + #startWork?: Promise; + #stopWork?: Promise; + #lossWork?: { generation: number; work: Promise }; + + 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 { + 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 { + 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 { + 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; + tracked = Promise.resolve().then(() => this.#start()).finally(() => { + if (this.#startWork === tracked) this.#startWork = undefined; + }); + this.#startWork = tracked; + return tracked; + } + + async #start(): Promise { + this.#running = true; + this.#reconnectAttempt = 0; + const generation = ++this.#generation; + this.#abort = new AbortController(); + await this.#connect(generation, false); + } + + public stop(): Promise { + if (this.#stopWork) return this.#stopWork; + if (!this.#running && !this.#startWork && this.#state.status === "stopped") { + return Promise.resolve(); + } + let tracked!: Promise; + tracked = Promise.resolve().then(() => this.#stop()).finally(() => { + if (this.#stopWork === tracked) this.#stopWork = undefined; + }); + this.#stopWork = tracked; + return tracked; + } + + async #stop(): Promise { + 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 { + 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 { + 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 { + 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 { + 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 { + 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 { + 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 { + 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 { + if (!this.#isCurrent(generation)) return; + const existing = this.#lossWork; + if (existing?.generation === generation) { + await existing.work; + return; + } + let tracked!: Promise; + 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 { + 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 { + 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 { + 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 { + try { + await this.#unsubscribe(); + } catch (error) { + this.#reportObserverError(error); + } + if (this.#isCurrent(generation)) this.#setState("degraded", message, this.#lease); + } + + async #unsubscribe(): Promise { + 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): Promise { + 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); +} diff --git a/packages/cultmesh-ts/test/provider-session.test.ts b/packages/cultmesh-ts/test/provider-session.test.ts new file mode 100644 index 00000000..8e1c9d50 --- /dev/null +++ b/packages/cultmesh-ts/test/provider-session.test.ts @@ -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(); + 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(); + 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(); + 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(); + 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(); + 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(); + + 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(); + 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; + }> = []; + + 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): 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 { + 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; + public registerError?: Error; + public publishBarrier?: Deferred; + public onFirstConnection?: (connection: FakeConnection) => void; + public readonly connections: FakeConnection[] = []; + + public constructor(private readonly scheduler: CultMeshProviderScheduler) {} + + public async connect( + connectedIdentity: CultMeshProviderIdentity, + signal: AbortSignal, + ): Promise { + 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; + public publishBarrier?: Deferred; + public unsubscribeError?: Error; + public withdrawError?: Error; + public closeError?: Error; + public closeAttempts = 0; + public afterPublishReceipt?: () => void | Promise; + public withdrawBarrier?: Deferred; + public closed = false; + #listener?: CultMeshProviderCommandListener; + #leaseSequence = 0; + + public constructor(private readonly scheduler: CultMeshProviderScheduler) {} + + public async register( + registration: CultMeshProviderRegistration, + signal: AbortSignal, + ): Promise { + this.registrations.push(registration); + if (this.registerBarrier) { + await Promise.race([ + this.registerBarrier.promise, + new Promise((_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 { + this.renewals.push(lease); + return this.lease(1_000); + } + + public async publish(publication: CultMeshProviderPublication): Promise { + 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 { + 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 { + 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 { + 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 { + 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 { + await new Promise(resolve => setImmediate(resolve)); +} + +class Deferred { + public readonly promise: Promise; + #resolve!: (value: T | PromiseLike) => void; + + public constructor() { + this.promise = new Promise(resolve => { this.#resolve = resolve; }); + } + + public resolve(value?: T): void { + this.#resolve(value as T); + } +} diff --git a/scripts/test-typescript-package-closure.mjs b/scripts/test-typescript-package-closure.mjs index 86334d53..c56a13a6 100644 --- a/scripts/test-typescript-package-closure.mjs +++ b/scripts/test-typescript-package-closure.mjs @@ -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"); diff --git a/src/GameCult.Mesh/docs/getting-started/04-keep-a-typescript-provider-alive.md b/src/GameCult.Mesh/docs/getting-started/04-keep-a-typescript-provider-alive.md new file mode 100644 index 00000000..15f10a3a --- /dev/null +++ b/src/GameCult.Mesh/docs/getting-started/04-keep-a-typescript-provider-alive.md @@ -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`. diff --git a/src/GameCult.Mesh/docs/getting-started/README.md b/src/GameCult.Mesh/docs/getting-started/README.md index ffb1094c..7914247c 100644 --- a/src/GameCult.Mesh/docs/getting-started/README.md +++ b/src/GameCult.Mesh/docs/getting-started/README.md @@ -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. - diff --git a/src/GameCult.Mesh/docs/reliability-migration-plan.md b/src/GameCult.Mesh/docs/reliability-migration-plan.md index dc50e7ef..c3b765a8 100644 --- a/src/GameCult.Mesh/docs/reliability-migration-plan.md +++ b/src/GameCult.Mesh/docs/reliability-migration-plan.md @@ -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