Repository Public
1/**2 * Outbound backpressure for a WebSocket fed by a pausable source (contract, "Terminal design"):3 * when the socket's buffered bytes exceed the high-water mark the source is paused; it is4 * resumed once the buffer has drained; if it stays saturated for `saturationMs` the5 * connection is declared dead (`onSaturated`, the caller closes with 4429).6 */7export interface BackpressureOptions {8 highWaterBytes: number;9 buffered: () => number;10 pause: () => void;11 resume: () => void;12 onSaturated: () => void;13 saturationMs?: number;14 pollMs?: number;15}1617export const SATURATION_MS = 30_000;1819export class Backpressure {20 private timer: NodeJS.Timeout | null = null;21 private pausedAt = 0;22 private disposed = false;23 private readonly saturationMs: number;24 private readonly pollMs: number;2526 constructor(private readonly o: BackpressureOptions) {27 this.saturationMs = o.saturationMs ?? SATURATION_MS;28 this.pollMs = o.pollMs ?? 50;29 }3031 get paused(): boolean {32 return this.timer !== null;33 }3435 /** Call after every send. */36 check(): void {37 if (this.disposed || this.timer !== null) return;38 if (this.o.buffered() <= this.o.highWaterBytes) return;39 this.pausedAt = Date.now();40 this.o.pause();41 this.timer = setInterval(() => this.poll(), this.pollMs);42 }4344 private poll(): void {45 if (this.disposed) return;46 // "Drained" = back under half the high-water mark, to avoid pause/resume flapping.47 if (this.o.buffered() <= this.o.highWaterBytes / 2) {48 this.clear();49 this.o.resume();50 return;51 }52 if (Date.now() - this.pausedAt >= this.saturationMs) {53 this.clear();54 this.disposed = true;55 this.o.onSaturated();56 }57 }5859 private clear(): void {60 if (this.timer) clearInterval(this.timer);61 this.timer = null;62 }6364 dispose(): void {65 this.disposed = true;66 this.clear();67 }68}69