From 40944dee17e2ca475f2b8d0ec3c7e32e3438162e Mon Sep 17 00:00:00 2001 From: Meta Vens Date: Wed, 23 Sep 2026 21:38:03 +0200 Subject: [PATCH] Cap the QUIC provider's inbound memory: publish-only mode and a connection cap Soul's probe proved a client can grow the provider's memory without limit (20k reliable frames plus 20k distinct latest-only keys took receiveQueueSize to 40000 and RSS from 97 to 318 MB) because StreamPixels never calls receive() to drain it, and there was no cap on connections either. The pinned certificate only authenticates the server, so any internet client can dial in. Add acceptClientFrames (default true, unchanged for existing callers and interop peers, which all use receive()): when false, any inbound stream a peer opens faults that connection at STREAM_STARTED, before a frame can be decoded or queued, so no inbox or generation-map entry is ever created for client data, and receive() itself rejects clearly. Add maxConnections (default 1024): a connection beyond the cap is refused at accept, shut down before any transport, outbox, or retained-frame seeding is allocated for it. Existing peers are unaffected. Co-Authored-By: Claude Sonnet 5 --- packages/cultmesh-ts/src/realtime-quic.ts | 60 ++++++++++++++++++++++- 1 file changed, 58 insertions(+), 2 deletions(-) diff --git a/packages/cultmesh-ts/src/realtime-quic.ts b/packages/cultmesh-ts/src/realtime-quic.ts index 210c2227..c368c96f 100644 --- a/packages/cultmesh-ts/src/realtime-quic.ts +++ b/packages/cultmesh-ts/src/realtime-quic.ts @@ -744,6 +744,28 @@ export interface CultMeshQuicRealtimeProviderOptions { * a DNS name. Defaults to `host`. */ readonly advertisedHost?: string; + /** + * When `false` (the default: every existing caller, test, and interop + * peer calls `receive()`), the provider fans client-originated streams + * into `receive()` as before. When `true`, the provider is publish-only: + * any stream a peer opens faults that connection at `STREAM_STARTED`, + * before a frame can be decoded or queued, so no inbox, generation-map, + * or fan-in entry is ever created for client data; `receive()` itself + * rejects. StreamPixels sets this explicitly — it closes the unbounded + * per-client memory growth Soul's probe found. + */ + readonly acceptClientFrames?: boolean; + /** + * Refuses a new connection at accept once the number of connections this + * provider is tracking (including mid-handshake ones) has reached this + * many: shut down immediately, before any transport, outbox, or + * retained-frame seeding is allocated for it. Existing peers are + * unaffected. Defaults to 1024 — far more than any single StreamPixels + * fan-out needs today, while still bounding per-connection overhead + * against a public UDP port whose pin authenticates only the server, so + * any internet client can dial in. StreamPixels sets its own value. + */ + readonly maxConnections?: number; } /** @@ -816,6 +838,8 @@ export class CultMeshQuicRealtimeProvider { * sweeps this set instead of only `peers` to close that gap. */ private readonly accepted = new Set(); + private readonly acceptClientFrames: boolean; + private readonly maxConnections: number; private disposed = false; private constructor( @@ -823,11 +847,15 @@ export class CultMeshQuicRealtimeProvider { listenerId: bigint, advertisedEndpoint: string, handshakeTimeoutMs: number, + acceptClientFrames: boolean, + maxConnections: number, ) { this.runtime = runtime; this.listenerId = listenerId; this.advertisedEndpoint = advertisedEndpoint; this.handshakeTimeoutMs = handshakeTimeoutMs; + this.acceptClientFrames = acceptClientFrames; + this.maxConnections = maxConnections; } /** Number of currently accepted peer connections. */ @@ -863,7 +891,14 @@ export class CultMeshQuicRealtimeProvider { } const advertisedEndpoint = `${CULTMESH_QUIC_REALTIME_SCHEME}://${formatEndpointHost(advertisedHost)}:${boundPort}?cert-sha256=${pinHex}`; - const provider = new CultMeshQuicRealtimeProvider(runtime, listenerId, advertisedEndpoint, handshakeTimeoutMs); + const provider = new CultMeshQuicRealtimeProvider( + runtime, + listenerId, + advertisedEndpoint, + handshakeTimeoutMs, + options.acceptClientFrames ?? true, + options.maxConnections ?? 1024, + ); runtime.onListenerEvent(listenerId, (event) => { if (event.type === CULTMESH_QUIC_EVENT_LISTENER_NEW_CONNECTION) provider.acceptConnection(event.connectionId); }); @@ -910,6 +945,11 @@ export class CultMeshQuicRealtimeProvider { /** The next client-originated frame, fanned in from every accepted peer. */ async receive(signal?: AbortSignal): Promise { + if (!this.acceptClientFrames) { + throw new Error( + "CultMesh QUIC realtime provider is publish-only (acceptClientFrames: false); it never accepts client frames.", + ); + } return this.received.receive(signal); } @@ -967,6 +1007,12 @@ export class CultMeshQuicRealtimeProvider { this.runtime.connectionShutdown(connectionId, CULTMESH_REALTIME_CONNECTION_CLOSE_CODE); return; } + if (this.accepted.size >= this.maxConnections) { + // Refused before any transport, timer, or outbox exists for it: + // existing accepted connections are untouched. + this.runtime.connectionShutdown(connectionId, CULTMESH_REALTIME_CONNECTION_CLOSE_CODE); + return; + } const transport = new CultMeshQuicRealtimeTransport( `quic-peer-${connectionId}`, @@ -992,6 +1038,16 @@ export class CultMeshQuicRealtimeProvider { this.runtime.onConnectionEvent( connectionId, (event) => { + // Publish-only: a peer-opened stream faults this connection alone, + // before `handleConnectionEvent` can record its stream kind or + // decode a frame off it. Throwing here drives the dispatch loop's + // own fault path (`CultMeshQuicNativeRuntime.dispatch`), which + // shuts the connection down and evicts it. + if (!this.acceptClientFrames && event.type === CULTMESH_QUIC_EVENT_STREAM_STARTED) { + throw new Error( + "CultMesh QUIC provider is publish-only (acceptClientFrames: false); it does not accept inbound streams from peers.", + ); + } transport.handleConnectionEvent(event); if (attached || event.type !== CULTMESH_QUIC_EVENT_CONNECTION_CONNECTED) return; attached = true; @@ -1003,7 +1059,7 @@ export class CultMeshQuicRealtimeProvider { for (const { frame, encoded } of this.retained.values()) outbox.publish(frame, encoded); this.outboxes.set(transport, outbox); this.peers.add(transport); - void this.pumpReceivedFrom(transport); + if (this.acceptClientFrames) void this.pumpReceivedFrom(transport); }, (error) => transport.fault(error), );