From fb5ee0d70821d0b22b649fc6f1848617b16a634b Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Tue, 1 Sep 2026 21:10:17 +0000 Subject: [PATCH 1/2] fix(core): stop dropping the first chunk of ffmpeg output when reading lines #readlines read one chunk before the loop and then immediately overwrote it with the next read, so every line in ffmpeg's first stderr chunk was silently discarded (and stderr that arrived in a single chunk produced no lines at all). Moves the generator to a module-level read_lines function so it can be unit tested directly. Co-authored-by: kaisea.rpi --- packages/core/src/lib/file_processor.ts | 40 +++++++++---------- packages/core/test/file_processor.test.ts | 48 +++++++++++++++++++++++ 2 files changed, 68 insertions(+), 20 deletions(-) create mode 100644 packages/core/test/file_processor.test.ts diff --git a/packages/core/src/lib/file_processor.ts b/packages/core/src/lib/file_processor.ts index f0b0347b..9fc151b2 100644 --- a/packages/core/src/lib/file_processor.ts +++ b/packages/core/src/lib/file_processor.ts @@ -513,7 +513,7 @@ class FileProcessor { let error_string = '' // TODO use the input timestamp in a progress meter - for await (const line of this.#readlines(proc.stderr)) { + for await (const line of read_lines(proc.stderr)) { if (line.includes('Parsed_showinfo') && line.includes(' n: ')) { const pts_time_str = line.match(/pts_time:(?\d+([.]\d*)?)/)?.groups?.pts_time if (pts_time_str === undefined) { @@ -648,29 +648,29 @@ class FileProcessor { }) } } +} - async * #readlines(stream: ReadableStream) { - const reader = stream.pipeThrough(new TextDecoderStream()).getReader() - let buffer = '' - let read_result = await reader.read() - while (!read_result.done) { - read_result = await reader.read() - buffer += read_result.value ?? '' - const lines = buffer.split('\n') - - // ignore the last line because we dont know where its newline is yet - for (let i = 0; i < lines.length - 1; i++) { - yield lines[i] - } - if (lines.length > 0) { - buffer = lines.at(-1)! - } +async function * read_lines(stream: ReadableStream) { + const reader = stream.pipeThrough(new TextDecoderStream()).getReader() + let buffer = '' + while (true) { + const read_result = await reader.read() + if (read_result.done) { + break } + buffer += read_result.value ?? '' + const lines = buffer.split('\n') - if (buffer.length > 0) { - yield buffer + // ignore the last line because we dont know where its newline is yet + for (let i = 0; i < lines.length - 1; i++) { + yield lines[i] } + buffer = lines.at(-1)! + } + + if (buffer.length > 0) { + yield buffer } } -export { FileProcessor } +export { FileProcessor, read_lines } diff --git a/packages/core/test/file_processor.test.ts b/packages/core/test/file_processor.test.ts new file mode 100644 index 00000000..a16ff2e2 --- /dev/null +++ b/packages/core/test/file_processor.test.ts @@ -0,0 +1,48 @@ +import { test } from 'forager-test' +import { read_lines } from '~/lib/file_processor.ts' + +function stream_of(chunks: string[]): ReadableStream { + const encoder = new TextEncoder() + return new ReadableStream({ + start(controller) { + for (const chunk of chunks) { + controller.enqueue(encoder.encode(chunk)) + } + controller.close() + } + }) +} + +test('read_lines', async (ctx) => { + await ctx.subtest('yields every line, including lines in the first chunk', async () => { + const lines = await Array.fromAsync(read_lines(stream_of([ + 'first\nsecond\n', + 'third\nfourth\n', + ]))) + ctx.assert.equals(lines, ['first', 'second', 'third', 'fourth']) + }) + + await ctx.subtest('reads a single chunk stream', async () => { + const lines = await Array.fromAsync(read_lines(stream_of(['only line\n']))) + ctx.assert.equals(lines, ['only line']) + }) + + await ctx.subtest('joins lines split across chunk boundaries', async () => { + const lines = await Array.fromAsync(read_lines(stream_of([ + 'frame ', + 'one\nframe ', + 'two', + ]))) + ctx.assert.equals(lines, ['frame one', 'frame two']) + }) + + await ctx.subtest('yields a trailing line without a newline', async () => { + const lines = await Array.fromAsync(read_lines(stream_of(['first\nno trailing newline']))) + ctx.assert.equals(lines, ['first', 'no trailing newline']) + }) + + await ctx.subtest('yields nothing for an empty stream', async () => { + const lines = await Array.fromAsync(read_lines(stream_of([]))) + ctx.assert.equals(lines, []) + }) +}) From 6fbf2c8b8cc482d6d871d2c05d1ce7f87f2f1559 Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Tue, 13 Oct 2026 16:01:03 +0000 Subject: [PATCH 2/2] refactor(core): make read_lines a static FileProcessor method Co-authored-by: kaisea.rpi --- packages/core/src/lib/file_processor.ts | 40 +++++++++++------------ packages/core/test/file_processor.test.ts | 14 ++++---- 2 files changed, 27 insertions(+), 27 deletions(-) diff --git a/packages/core/src/lib/file_processor.ts b/packages/core/src/lib/file_processor.ts index 9fc151b2..a1471518 100644 --- a/packages/core/src/lib/file_processor.ts +++ b/packages/core/src/lib/file_processor.ts @@ -513,7 +513,7 @@ class FileProcessor { let error_string = '' // TODO use the input timestamp in a progress meter - for await (const line of read_lines(proc.stderr)) { + for await (const line of FileProcessor.read_lines(proc.stderr)) { if (line.includes('Parsed_showinfo') && line.includes(' n: ')) { const pts_time_str = line.match(/pts_time:(?\d+([.]\d*)?)/)?.groups?.pts_time if (pts_time_str === undefined) { @@ -648,29 +648,29 @@ class FileProcessor { }) } } -} -async function * read_lines(stream: ReadableStream) { - const reader = stream.pipeThrough(new TextDecoderStream()).getReader() - let buffer = '' - while (true) { - const read_result = await reader.read() - if (read_result.done) { - break - } - buffer += read_result.value ?? '' - const lines = buffer.split('\n') + static async * read_lines(stream: ReadableStream) { + const reader = stream.pipeThrough(new TextDecoderStream()).getReader() + let buffer = '' + while (true) { + const read_result = await reader.read() + if (read_result.done) { + break + } + buffer += read_result.value ?? '' + const lines = buffer.split('\n') - // ignore the last line because we dont know where its newline is yet - for (let i = 0; i < lines.length - 1; i++) { - yield lines[i] + // ignore the last line because we dont know where its newline is yet + for (let i = 0; i < lines.length - 1; i++) { + yield lines[i] + } + buffer = lines.at(-1)! } - buffer = lines.at(-1)! - } - if (buffer.length > 0) { - yield buffer + if (buffer.length > 0) { + yield buffer + } } } -export { FileProcessor, read_lines } +export { FileProcessor } diff --git a/packages/core/test/file_processor.test.ts b/packages/core/test/file_processor.test.ts index a16ff2e2..b6ce296d 100644 --- a/packages/core/test/file_processor.test.ts +++ b/packages/core/test/file_processor.test.ts @@ -1,5 +1,5 @@ import { test } from 'forager-test' -import { read_lines } from '~/lib/file_processor.ts' +import { FileProcessor } from '~/lib/file_processor.ts' function stream_of(chunks: string[]): ReadableStream { const encoder = new TextEncoder() @@ -13,9 +13,9 @@ function stream_of(chunks: string[]): ReadableStream { }) } -test('read_lines', async (ctx) => { +test('FileProcessor.read_lines', async (ctx) => { await ctx.subtest('yields every line, including lines in the first chunk', async () => { - const lines = await Array.fromAsync(read_lines(stream_of([ + const lines = await Array.fromAsync(FileProcessor.read_lines(stream_of([ 'first\nsecond\n', 'third\nfourth\n', ]))) @@ -23,12 +23,12 @@ test('read_lines', async (ctx) => { }) await ctx.subtest('reads a single chunk stream', async () => { - const lines = await Array.fromAsync(read_lines(stream_of(['only line\n']))) + const lines = await Array.fromAsync(FileProcessor.read_lines(stream_of(['only line\n']))) ctx.assert.equals(lines, ['only line']) }) await ctx.subtest('joins lines split across chunk boundaries', async () => { - const lines = await Array.fromAsync(read_lines(stream_of([ + const lines = await Array.fromAsync(FileProcessor.read_lines(stream_of([ 'frame ', 'one\nframe ', 'two', @@ -37,12 +37,12 @@ test('read_lines', async (ctx) => { }) await ctx.subtest('yields a trailing line without a newline', async () => { - const lines = await Array.fromAsync(read_lines(stream_of(['first\nno trailing newline']))) + const lines = await Array.fromAsync(FileProcessor.read_lines(stream_of(['first\nno trailing newline']))) ctx.assert.equals(lines, ['first', 'no trailing newline']) }) await ctx.subtest('yields nothing for an empty stream', async () => { - const lines = await Array.fromAsync(read_lines(stream_of([]))) + const lines = await Array.fromAsync(FileProcessor.read_lines(stream_of([]))) ctx.assert.equals(lines, []) }) })