From 04858d332a6bdd35a43833f82d0a805eb4a04261 Mon Sep 17 00:00:00 2001 From: Meta Vens Date: Wed, 23 Sep 2026 14:38:34 +0200 Subject: [PATCH] Fix the QUIC consumer's peer-shutdown leak, and give the session manager eviction, single-flight and first-success racing Soul's fix batch, findings 3, 4, 5 and 6: - A peer-initiated CONNECTION_SHUTDOWN only set `disposed = true`, so `dispose()`'s early-return skipped `release()` and `offConnectionEvent`: the runtime reference leaked forever and a pending `receiveFrame` hung forever. `disposed` and "already cleaned up" are now the same flag, run through one `cleanup()` reached from `dispose()`, a peer shutdown, and the new `fault()` (wired from fix 1's `onFault`). Every path releases the runtime reference exactly once and rejects every pending and future receive/send with the real reason instead of a generic "disposed" message. - `CultMeshQuicRealtimeSessionManager.connect()` cached a transport forever, even a dead one, and raced candidates with `Promise.allSettled`, so one dead route delayed every connect by the full handshake timeout. It now: evicts a session once its transport reports itself disposed (`onDisposed`, mirroring `CultMeshSessions.InvalidateRealtimeSession`); shares one in-flight dial across concurrent callers for the same target (mirrors C#'s `Lazy`); refuses a transport that lands after `dispose()` instead of caching it; and races each tier for the first verified success (mirrors `Task.WhenAny`), aborting and disposing the losers via an `AbortSignal` threaded into `connector.connect`, which now also honours a signal already aborted at entry. - `streamKinds` never pruned a finished stream (`STREAM_SHUTDOWN` fell into `default`); it now removes the entry. The receive side queued every latest-only frame in an unbounded array; it now mirrors the C# reference's `CultMeshRealtimeInbox`: at most one pending frame per `(channel, body)`, coalesced on arrival, while reliable-ordered frames still queue directly (unbounded, matching the C# reference's own unbounded channel). Subtraction: `trimmedOrEmpty` duplicated `cultnet-ts`'s exported `trimCSharp`; callers now use the shared one. Co-Authored-By: Claude Sonnet 5 --- packages/cultmesh-ts/src/realtime-quic.ts | 351 +++++++++++++++++----- 1 file changed, 279 insertions(+), 72 deletions(-) diff --git a/packages/cultmesh-ts/src/realtime-quic.ts b/packages/cultmesh-ts/src/realtime-quic.ts index b1468290..5ddf265f 100644 --- a/packages/cultmesh-ts/src/realtime-quic.ts +++ b/packages/cultmesh-ts/src/realtime-quic.ts @@ -16,7 +16,7 @@ import { createHash } from "node:crypto"; import { - isCSharpWhiteSpace, + trimCSharp, verifyAuthorityRoute, type CultMeshAuthorityRouteCertificate, type CultMeshAuthorityTrustPolicy, @@ -40,6 +40,7 @@ import { CULTMESH_QUIC_EVENT_CONNECTION_SHUTDOWN, CULTMESH_QUIC_EVENT_STREAM_FRAME, CULTMESH_QUIC_EVENT_STREAM_SEND_COMPLETE, + CULTMESH_QUIC_EVENT_STREAM_SHUTDOWN, CULTMESH_QUIC_EVENT_STREAM_STARTED, CULTMESH_QUIC_SEND_CANCELED, CULTMESH_QUIC_STREAM_LATEST_ONLY, @@ -81,6 +82,13 @@ export interface CultMeshRealtimeTransport { protocolId: string, routeGeneration: string, ): boolean; + /** + * Registers a handler invoked once this transport becomes unusable for any + * reason (explicit `dispose()`, a peer-initiated shutdown, or an internal + * fault). A session manager uses this to evict a dead cached session so the + * next `connect()` redials, mirroring `CultMeshSessions.InvalidateRealtimeSession`. + */ + onDisposed?(handler: () => void): void; } /** Creates realtime transports; the TS analogue of `ICultMeshRealtimeTransportConnector`. */ @@ -133,6 +141,14 @@ interface PendingSend { reject(error: Error): void; } +interface PendingReceive { + resolve(frame: CultMeshRealtimeFrame): void; + reject(error: Error): void; +} + +/** A direct frame, or a `latestPending` key to resolve at delivery time. */ +type ReadyToken = { readonly frame: CultMeshRealtimeFrame } | { readonly latestKey: string }; + interface LatestGeneration { producerEpoch: bigint; sequence: bigint; @@ -194,6 +210,9 @@ export class CultMeshQuicRealtimeConnector implements CultMeshRealtimeTransportC target: CultMeshRealtimeTarget, signal?: AbortSignal, ): Promise { + if (signal?.aborted) { + throw new Error("CultMesh QUIC connect aborted."); + } const { host, port, certificateSha256 } = parseQuicRealtimeEndpoint(candidate.endpoint); const runtime = await CultMeshQuicNativeRuntime.open(); let connectionId: bigint; @@ -209,6 +228,16 @@ export class CultMeshQuicRealtimeConnector implements CultMeshRealtimeTransportC return await new Promise((resolve, reject) => { let settled = false; + + // Honour a signal already aborted by the time we reach here (it fired + // before `addEventListener` could observe it, so it never would). + if (signal?.aborted) { + runtime.connectionShutdown(connectionId, CULTMESH_REALTIME_CONNECTION_CLOSE_CODE); + void runtime.release(); + reject(new Error("CultMesh QUIC connect aborted.")); + return; + } + const timer = setTimeout(() => { if (settled) return; settled = true; @@ -242,6 +271,18 @@ export class CultMeshQuicRealtimeConnector implements CultMeshRealtimeTransportC resolve(transport!); }; + // Told by the runtime when this connection faults mid-dispatch (a + // throwing validator during the handshake, or a malformed frame / + // stream-kind mismatch afterwards): reject the still-pending connect, + // or hand the fault to the connected transport. + const onFault = (error: Error): void => { + if (!settled) { + finish(undefined, error); + return; + } + transport.fault(error); + }; + const transport = new CultMeshQuicRealtimeTransport( candidate.endpoint, runtime, @@ -250,33 +291,37 @@ export class CultMeshQuicRealtimeConnector implements CultMeshRealtimeTransportC candidate.generation, ); - runtime.onConnectionEvent(connectionId, (event) => { - if (event.type === CULTMESH_QUIC_EVENT_CONNECTION_CERTIFICATE_RECEIVED) { - const der = event.payload; - const accepted = validator - ? validator(target, der) - : certificateSha256 !== undefined && - certificateSha256Hex(der).toLowerCase() === certificateSha256.toLowerCase(); - try { - runtime.connectionCertificateComplete(connectionId, accepted); - } catch (error) { - finish(undefined, error instanceof Error ? error : new Error(String(error))); + runtime.onConnectionEvent( + connectionId, + (event) => { + if (event.type === CULTMESH_QUIC_EVENT_CONNECTION_CERTIFICATE_RECEIVED) { + const der = event.payload; + const accepted = validator + ? validator(target, der) + : certificateSha256 !== undefined && + certificateSha256Hex(der).toLowerCase() === certificateSha256.toLowerCase(); + try { + runtime.connectionCertificateComplete(connectionId, accepted); + } catch (error) { + finish(undefined, error instanceof Error ? error : new Error(String(error))); + } + if (!accepted) { + finish(undefined, new Error(`CultMesh QUIC provider certificate for '${candidate.endpoint}' was rejected.`)); + } + return; } - if (!accepted) { - finish(undefined, new Error(`CultMesh QUIC provider certificate for '${candidate.endpoint}' was rejected.`)); + if (event.type === CULTMESH_QUIC_EVENT_CONNECTION_SHUTDOWN && !settled) { + const reason = Buffer.from(event.payload).toString("utf8"); + finish(undefined, new Error(`CultMesh QUIC connection closed during handshake: ${reason || event.code}`)); + return; } - return; - } - if (event.type === CULTMESH_QUIC_EVENT_CONNECTION_SHUTDOWN && !settled) { - const reason = Buffer.from(event.payload).toString("utf8"); - finish(undefined, new Error(`CultMesh QUIC connection closed during handshake: ${reason || event.code}`)); - return; - } - transport.handleConnectionEvent(event); - if (!settled && event.type === CULTMESH_QUIC_EVENT_CONNECTION_CONNECTED) { - finish(transport); - } - }); + transport.handleConnectionEvent(event); + if (!settled && event.type === CULTMESH_QUIC_EVENT_CONNECTION_CONNECTED) { + finish(transport); + } + }, + onFault, + ); }); } } @@ -298,11 +343,20 @@ export class CultMeshQuicRealtimeTransport implements CultMeshRealtimeTransport private readonly reliableGate = new SimpleGate(); private readonly pendingSends = new Map(); private readonly streamKinds = new Map(); + /** Generation staleness filter: the newest `(channel, body)` generation delivered so far. */ private readonly latestGenerations = new Map(); - private readonly receiveQueue: CultMeshRealtimeFrame[] = []; - private readonly receiveWaiters: Array<(frame: CultMeshRealtimeFrame) => void> = []; + /** Inbox coalescing (mirrors `CultMeshRealtimeInbox`): the newest not-yet-delivered + * latest-only frame per `(channel, body)`, so a slow reader accumulates at most one + * pending frame per key instead of an unbounded backlog. */ + private readonly latestPending = new Map(); + /** FIFO of ready tokens: a direct frame (reliable-ordered/unreliable), or a + * `(channel, body)` key to look up in `latestPending` at delivery time. */ + private readonly readyTokens: ReadyToken[] = []; + private readonly receiveWaiters: PendingReceive[] = []; private reliableOutboundStreamId: bigint | undefined; private disposed = false; + private terminalError: Error | undefined; + private readonly disposalHandlers = new Set<() => void>(); constructor( endpoint: string, @@ -372,34 +426,80 @@ export class CultMeshQuicRealtimeTransport implements CultMeshRealtimeTransport } async receiveFrame(signal?: AbortSignal): Promise { - if (this.receiveQueue.length > 0) return this.receiveQueue.shift()!; - if (this.disposed) throw new Error("CultMesh QUIC realtime transport is disposed."); + if (this.readyTokens.length > 0) return this.resolveToken(this.readyTokens.shift()!); + if (this.disposed) throw this.terminalError ?? new Error("CultMesh QUIC realtime transport is disposed."); return await new Promise((resolve, reject) => { const onAbort = (): void => { const index = this.receiveWaiters.indexOf(waiter); if (index >= 0) this.receiveWaiters.splice(index, 1); reject(new Error("CultMesh QUIC receiveFrame aborted.")); }; - const waiter = (frame: CultMeshRealtimeFrame): void => { - signal?.removeEventListener("abort", onAbort); - resolve(frame); + const waiter: PendingReceive = { + resolve: (frame) => { + signal?.removeEventListener("abort", onAbort); + resolve(frame); + }, + reject: (error) => { + signal?.removeEventListener("abort", onAbort); + reject(error); + }, }; signal?.addEventListener("abort", onAbort, { once: true }); this.receiveWaiters.push(waiter); }); } + /** Registers a handler invoked once, when this transport becomes unusable. */ + onDisposed(handler: () => void): void { + if (this.disposed) { + handler(); + return; + } + this.disposalHandlers.add(handler); + } + dispose(): void { if (this.disposed) return; - this.disposed = true; - this.runtime.offConnectionEvent(this.connectionId); - for (const [, pending] of this.pendingSends) pending.reject(new Error("CultMesh QUIC realtime transport is disposed.")); - this.pendingSends.clear(); try { this.runtime.connectionShutdown(this.connectionId, CULTMESH_REALTIME_CONNECTION_CLOSE_CODE); } catch { // Best-effort: the connection may already be gone. } + this.cleanup(); + } + + /** + * @internal Called by the connector when the native runtime faults this + * connection mid-dispatch (a malformed frame, a stream-kind mismatch, or a + * throwing caller callback). Mirrors the C# reference's + * `_received.Complete(error)` / `_shutdown.Cancel()`: pending and future + * `receiveFrame` calls reject with `error`, and the connection is torn + * down. The native side has already shut the connection down and removed + * its listener by the time this runs. + */ + fault(error: Error): void { + this.cleanup(error); + } + + /** + * Runs exactly once, however the transport becomes unusable: explicit + * `dispose()`, a peer-initiated `CONNECTION_SHUTDOWN`, or `fault()`. + * Releases the runtime reference and rejects every pending and future + * receive/send, so none of those paths can leak the reference or hang a + * waiter forever. + */ + private cleanup(error?: Error): void { + if (this.disposed) return; + this.disposed = true; + this.terminalError = error; + this.runtime.offConnectionEvent(this.connectionId); + const rejection = error ?? new Error("CultMesh QUIC realtime transport is disposed."); + for (const [, pending] of this.pendingSends) pending.reject(rejection); + this.pendingSends.clear(); + const waiters = this.receiveWaiters.splice(0, this.receiveWaiters.length); + for (const waiter of waiters) waiter.reject(rejection); + for (const handler of this.disposalHandlers) handler(); + this.disposalHandlers.clear(); void this.runtime.release(); } @@ -412,6 +512,9 @@ export class CultMeshQuicRealtimeTransport implements CultMeshRealtimeTransport case CULTMESH_QUIC_EVENT_STREAM_FRAME: this.onStreamFrame(event); return; + case CULTMESH_QUIC_EVENT_STREAM_SHUTDOWN: + this.streamKinds.delete(event.streamId); + return; case CULTMESH_QUIC_EVENT_STREAM_SEND_COMPLETE: { const pending = this.pendingSends.get(event.streamId); if (!pending) return; @@ -424,14 +527,19 @@ export class CultMeshQuicRealtimeTransport implements CultMeshRealtimeTransport return; } case CULTMESH_QUIC_EVENT_CONNECTION_SHUTDOWN: - this.disposed = true; + this.cleanup(new Error("CultMesh QUIC connection was closed by the remote peer.")); return; default: return; } } - /** `ReadStreamAsync`'s stream-kind/delivery consistency check (`:519-524`). */ + /** + * `ReadStreamAsync`'s stream-kind/delivery consistency check (`:519-524`). + * Throwing here is intentional: the native runtime's dispatch loop (fix 1) + * catches it, faults only this connection, and keeps the pump and every + * other connection running. + */ private onStreamFrame(event: CultMeshQuicNativeEvent): void { const kind = this.streamKinds.get(event.streamId); const frame = decodeRealtimeFrame(event.payload); @@ -444,7 +552,14 @@ export class CultMeshQuicRealtimeTransport implements CultMeshRealtimeTransport this.publishReceived(frame); } - /** `PublishReceived`'s per-`(channel, body)` generation filter (`:538-560`). */ + /** + * `PublishReceived`'s per-`(channel, body)` generation filter (`:538-560`), + * plus the inbox's own coalescing (`CultMeshRealtimeInbox.Publish`, + * `:580-595`): a latest-only frame overwrites the still-pending frame for + * its key rather than queuing beside it, so an unread backlog holds at most + * one frame per `(channel, body)`. Reliable-ordered frames queue directly, + * unbounded, exactly as the C# reference's unbounded channel does. + */ private publishReceived(frame: CultMeshRealtimeFrame): void { if (frame.delivery === "latest-only") { const key = frame.channelId + "\u001f" + frame.bodyId; @@ -452,13 +567,32 @@ export class CultMeshQuicRealtimeTransport implements CultMeshRealtimeTransport const current = this.latestGenerations.get(key); if (current && compareGeneration(candidate, current) <= 0) return; this.latestGenerations.set(key, candidate); + + const alreadyPending = this.latestPending.has(key); + this.latestPending.set(key, frame); + if (alreadyPending) return; + this.enqueueReady({ latestKey: key }); + return; } + this.enqueueReady({ frame }); + } + + private enqueueReady(token: ReadyToken): void { const waiter = this.receiveWaiters.shift(); if (waiter) { - waiter(frame); - } else { - this.receiveQueue.push(frame); + waiter.resolve(this.resolveToken(token)); + return; } + this.readyTokens.push(token); + } + + private resolveToken(token: ReadyToken): CultMeshRealtimeFrame { + if ("frame" in token) return token.frame; + const frame = this.latestPending.get(token.latestKey); + this.latestPending.delete(token.latestKey); + // Always present: a key is only ever enqueued alongside setting it, and + // removed here on the one dequeue that follows. + return frame!; } } @@ -493,14 +627,6 @@ function routeCertificateView( }; } -function trimmedOrEmpty(value: string): string { - let start = 0; - let end = value.length; - while (start < end && isCSharpWhiteSpace(value[start]!)) start += 1; - while (end > start && isCSharpWhiteSpace(value[end - 1]!)) end -= 1; - return value.slice(start, end); -} - /** Options for `CultMeshQuicRealtimeSessionManager`. */ export interface CultMeshQuicRealtimeSessionManagerOptions { readonly lookupSource: ICultMeshRealtimeLookupSource; @@ -517,6 +643,9 @@ export interface CultMeshQuicRealtimeSessionManagerOptions { export class CultMeshQuicRealtimeSessionManager { private readonly options: CultMeshQuicRealtimeSessionManagerOptions; private readonly sessions = new Map(); + /** Single-flight: concurrent `connect()` calls for the same key share one dial. */ + private readonly connecting = new Map>(); + private disposed = false; constructor(options: CultMeshQuicRealtimeSessionManagerOptions) { if (options.connectors.length === 0) { @@ -532,12 +661,44 @@ export class CultMeshQuicRealtimeSessionManager { return target.verseId + "\u001f" + target.authorityRuntimeId; } - /** Resolves, verifies, races and connects; disposes losing candidates. */ + /** + * Resolves, verifies, races and connects; disposes losing candidates. + * Returns the cached session for `target` when one is live. Concurrent + * calls for the same target share one dial (mirrors C#'s `Lazy`); a + * transport that dies later (peer shutdown, an internal fault, or explicit + * disposal) evicts itself from the cache so the next call redials, mirroring + * `CultMeshSessions.InvalidateRealtimeSession` (`:898-913`). + */ async connect(target: CultMeshRealtimeTarget): Promise { + if (this.disposed) throw new Error("CultMeshQuicRealtimeSessionManager is disposed."); const key = this.sessionKey(target); const existing = this.sessions.get(key); if (existing) return existing; + const inFlight = this.connecting.get(key); + if (inFlight) return inFlight; + + const attempt = this.connectOwnedAsync(key, target).finally(() => { + this.connecting.delete(key); + }); + this.connecting.set(key, attempt); + return attempt; + } + + private async connectOwnedAsync(key: string, target: CultMeshRealtimeTarget): Promise { + const transport = await this.dialAsync(key, target); + if (this.disposed) { + transport.dispose(); + throw new Error("CultMeshQuicRealtimeSessionManager was disposed while connecting."); + } + this.sessions.set(key, transport); + transport.onDisposed?.(() => { + if (this.sessions.get(key) === transport) this.sessions.delete(key); + }); + return transport; + } + + private async dialAsync(key: string, target: CultMeshRealtimeTarget): Promise { const verses = await this.options.lookupSource.resolve(target); const routes = verses .flatMap((verse) => verse.authorityRoutes ?? []) @@ -554,7 +715,7 @@ export class CultMeshQuicRealtimeSessionManager { endpoint: route.endpoint, authorityRuntimeId: route.authorityRuntimeId, priority: route.priority, - generation: trimmedOrEmpty(route.generation), + generation: trimCSharp(route.generation), authorityRoute: route, }; try { @@ -601,31 +762,76 @@ export class CultMeshQuicRealtimeSessionManager { const entries = [...(tiers.get(tier) ?? [])] .sort((a, b) => a.candidate.priority - b.candidate.priority) .slice(0, maxRaced); - const attempts = entries.map(async (entry) => { - const transport = await entry.connector.connect(entry.candidate, target); - if (!transport.isVerifiedFor(target.verseId, target.authorityRuntimeId, CULTMESH_REALTIME_STATE_PROTOCOL_ID, entry.candidate.generation)) { - transport.dispose(); - throw new Error(`Transport did not prove CultMesh authority for '${key}'.`); - } - return transport; - }); - const settled = await Promise.allSettled(attempts); - const winner = settled.find((result): result is PromiseFulfilledResult => result.status === "fulfilled"); - for (const result of settled) { - if (result === winner) continue; - if (result.status === "fulfilled") result.value.dispose(); - else failures.push(result.reason instanceof Error ? result.reason : new Error(String(result.reason))); - } - if (winner) { - this.sessions.set(key, winner.value); - return winner.value; - } + const winner = await this.raceFirstSuccessAsync(key, target, entries, failures); + if (winner) return winner; } throw new Error( `No realtime state path connected for '${key}': ${failures.map((error) => error.message).join("; ")}`, ); } + /** + * Races one tier's candidates and returns the first verified success, + * mirroring C#'s `Task.WhenAny` loop (`:851-866`): as soon as one candidate + * both connects and proves `isVerifiedFor`, the rest are aborted (an + * `AbortSignal` is threaded into `connector.connect`) and disposed once + * they settle, without blocking the return. `Promise.allSettled` here would + * instead wait out every candidate, including a dead route stalled on its + * own handshake timeout. + */ + private async raceFirstSuccessAsync( + key: string, + target: CultMeshRealtimeTarget, + entries: ReadonlyArray<{ candidate: CultMeshRealtimeCandidate; connector: CultMeshRealtimeTransportConnector }>, + failures: Error[], + ): Promise { + const controller = new AbortController(); + const pending = new Set( + entries.map((entry) => this.connectCandidateAsync(key, target, entry, controller.signal)), + ); + let winner: CultMeshRealtimeTransport | undefined; + while (pending.size > 0 && !winner) { + const settled = await Promise.race( + [...pending].map((attempt) => + attempt.then( + (value) => ({ attempt, ok: true as const, value }), + (error) => ({ attempt, ok: false as const, error }), + ), + ), + ); + pending.delete(settled.attempt); + if (settled.ok) { + winner = settled.value; + } else { + failures.push(settled.error instanceof Error ? settled.error : new Error(String(settled.error))); + } + } + if (!winner) return undefined; + + controller.abort(); + // The remaining candidates may already be in flight past the point an + // abort signal helps; dispose whatever they eventually produce, without + // blocking the winner's return. + for (const loser of pending) { + loser.then((transport) => transport.dispose()).catch(() => {}); + } + return winner; + } + + private async connectCandidateAsync( + key: string, + target: CultMeshRealtimeTarget, + entry: { candidate: CultMeshRealtimeCandidate; connector: CultMeshRealtimeTransportConnector }, + signal: AbortSignal, + ): Promise { + const transport = await entry.connector.connect(entry.candidate, target, signal); + if (!transport.isVerifiedFor(target.verseId, target.authorityRuntimeId, CULTMESH_REALTIME_STATE_PROTOCOL_ID, entry.candidate.generation)) { + transport.dispose(); + throw new Error(`Transport did not prove CultMesh authority for '${key}'.`); + } + return transport; + } + /** Marks the session for `target` offline, disposing its transport. */ disconnect(target: CultMeshRealtimeTarget): void { const key = this.sessionKey(target); @@ -636,6 +842,7 @@ export class CultMeshQuicRealtimeSessionManager { } dispose(): void { + this.disposed = true; for (const transport of this.sessions.values()) transport.dispose(); this.sessions.clear(); }