(body: ReadableStream<Uint8Array>)
| 337 | */ |
| 338 | // eslint-disable-next-line eslint-plugin-n/no-unsupported-features/node-builtins |
| 339 | private async readStream(body: ReadableStream<Uint8Array>): Promise<void> { |
| 340 | const reader = body.getReader() |
| 341 | const decoder = new TextDecoder() |
| 342 | let buffer = '' |
| 343 | |
| 344 | try { |
| 345 | while (true) { |
| 346 | const { done, value } = await reader.read() |
| 347 | if (done) break |
| 348 | |
| 349 | buffer += decoder.decode(value, STREAM_DECODE_OPTS) |
| 350 | const { frames, remaining } = parseSSEFrames(buffer) |
| 351 | buffer = remaining |
| 352 | |
| 353 | for (const frame of frames) { |
| 354 | // Any frame (including keepalive comments) proves the connection is alive |
| 355 | this.resetLivenessTimer() |
| 356 | |
| 357 | if (frame.id) { |
| 358 | const seqNum = parseInt(frame.id, 10) |
| 359 | if (!isNaN(seqNum)) { |
| 360 | if (this.seenSequenceNums.has(seqNum)) { |
| 361 | logForDebugging( |
| 362 | `SSETransport: DUPLICATE frame seq=${seqNum} (lastSequenceNum=${this.lastSequenceNum}, seenCount=${this.seenSequenceNums.size})`, |
| 363 | { level: 'warn' }, |
| 364 | ) |
| 365 | logForDiagnosticsNoPII('warn', 'cli_sse_duplicate_sequence') |
| 366 | } else { |
| 367 | this.seenSequenceNums.add(seqNum) |
| 368 | // Prevent unbounded growth: once we have many entries, prune |
| 369 | // old sequence numbers that are well below the high-water mark. |
| 370 | // Only sequence numbers near lastSequenceNum matter for dedup. |
| 371 | if (this.seenSequenceNums.size > 1000) { |
| 372 | const threshold = this.lastSequenceNum - 200 |
| 373 | for (const s of this.seenSequenceNums) { |
| 374 | if (s < threshold) { |
| 375 | this.seenSequenceNums.delete(s) |
| 376 | } |
| 377 | } |
| 378 | } |
| 379 | } |
| 380 | if (seqNum > this.lastSequenceNum) { |
| 381 | this.lastSequenceNum = seqNum |
| 382 | } |
| 383 | } |
| 384 | } |
| 385 | |
| 386 | if (frame.event && frame.data) { |
| 387 | this.handleSSEFrame(frame.event, frame.data) |
| 388 | } else if (frame.data) { |
| 389 | // data: without event: — server is emitting the old envelope format |
| 390 | // or a bug. Log so incidents show as a signal instead of silent drops. |
| 391 | logForDebugging( |
| 392 | 'SSETransport: Frame has data: but no event: field — dropped', |
| 393 | { level: 'warn' }, |
| 394 | ) |
| 395 | logForDiagnosticsNoPII('warn', 'cli_sse_frame_missing_event_field') |
| 396 | } |
no test coverage detected