3 Commits

Author SHA1 Message Date
ea234ea007 Merge pull request 'fix: yield to event loop after each SSE write to flush socket' (#10) from fix/sse-event-loop-flush into main
Reviewed-on: #10
2026-03-14 17:00:51 +00:00
vakabunga
db4d5e4d00 fix: yield to event loop after each SSE write to flush socket
The for-await loop over OpenAI stream chunks runs synchronously when
data is buffered, causing res.write() calls to queue without flushing.
Add setImmediate yield after each progress event so the event loop
reaches its I/O phase and pushes data to the network immediately.
2026-03-14 19:59:22 +03:00
358fcaeff5 Merge pull request 'fix: disable gzip and pad SSE events to prevent proxy buffering' (#9) from fix/sse-gzip-buffering into main
Reviewed-on: #9
2026-03-14 16:46:07 +00:00
2 changed files with 10 additions and 8 deletions

View File

@@ -28,14 +28,8 @@ function isJsonFile(file: { mimetype: string; originalname: string }): boolean {
); );
} }
const SSE_PAD_TARGET = 4096;
function sseWrite(res: import('express').Response, data: Record<string, unknown>) { function sseWrite(res: import('express').Response, data: Record<string, unknown>) {
const payload = `data: ${JSON.stringify(data)}\n\n`; res.write(`data: ${JSON.stringify(data)}\n\n`);
const pad = Math.max(0, SSE_PAD_TARGET - payload.length);
// SSE comment lines (": ...") are ignored by the browser but push
// data past proxy buffer thresholds so each event is delivered immediately.
res.write(pad > 0 ? `: ${' '.repeat(pad)}\n${payload}` : payload);
} }
const router = Router(); const router = Router();

View File

@@ -150,6 +150,10 @@ const LLM_PROGRESS_MAX = 98;
const LLM_PROGRESS_RANGE = LLM_PROGRESS_MAX - LLM_PROGRESS_MIN; const LLM_PROGRESS_RANGE = LLM_PROGRESS_MAX - LLM_PROGRESS_MIN;
const THROTTLE_MS = 300; const THROTTLE_MS = 300;
function yieldToEventLoop(): Promise<void> {
return new Promise(resolve => setImmediate(resolve));
}
export async function convertPdfToStatementStreaming( export async function convertPdfToStatementStreaming(
buffer: Buffer, buffer: Buffer,
onProgress: OnProgress, onProgress: OnProgress,
@@ -163,6 +167,7 @@ export async function convertPdfToStatementStreaming(
} }
onProgress('pdf', 2, 'Извлечение текста из PDF...'); onProgress('pdf', 2, 'Извлечение текста из PDF...');
await yieldToEventLoop();
let text: string; let text: string;
try { try {
@@ -186,6 +191,7 @@ export async function convertPdfToStatementStreaming(
} }
onProgress('pdf', 8, 'Текст извлечён, отправка в LLM...'); onProgress('pdf', 8, 'Текст извлечён, отправка в LLM...');
await yieldToEventLoop();
const openai = new OpenAI({ const openai = new OpenAI({
apiKey: config.llmApiKey, apiKey: config.llmApiKey,
@@ -205,7 +211,6 @@ export async function convertPdfToStatementStreaming(
stream: true, stream: true,
}); });
// Estimate expected output size as ~2x the input PDF text length, clamped
const expectedChars = Math.max(2_000, Math.min(text.length * 2, 30_000)); const expectedChars = Math.max(2_000, Math.min(text.length * 2, 30_000));
let accumulated = ''; let accumulated = '';
@@ -227,11 +232,14 @@ export async function convertPdfToStatementStreaming(
); );
onProgress('llm', llmProgress, 'Конвертация через LLM...'); onProgress('llm', llmProgress, 'Конвертация через LLM...');
lastEmitTime = now; lastEmitTime = now;
// Let the event loop flush socket writes to the network
await yieldToEventLoop();
} }
} }
} }
onProgress('llm', LLM_PROGRESS_MAX, 'LLM завершил, обработка результата...'); onProgress('llm', LLM_PROGRESS_MAX, 'LLM завершил, обработка результата...');
await yieldToEventLoop();
const content = accumulated.trim(); const content = accumulated.trim();
if (!content) { if (!content) {