From eb456d25f01611c1e1cb55368ccc133c211a0718 Mon Sep 17 00:00:00 2001 From: CyberWolf Date: Thu, 5 Nov 2026 04:12:54 +0800 Subject: [PATCH] fix(split): render PDF pages in batches and stop split crash loops (#88) pdfToPng holds every page it renders until it returns, so a large PDF kept all its PNGs in the main process. Pages are now rendered and written in batches of WORKER_CONFIG.splitter.renderBatchSize, and only their metadata is kept. Without a page range, pdf.js still decides where the document ends, because pdf-lib misses page tree nodes that lack a /Type. A split that brought the app down was re-queued on every start. Interrupted splits are now counted in Task.split_interrupt_count, and a task fails once it reaches maxInterruptedSplits. Retrying the task resets the count. --- .../services/WorkerOrchestrator.ts | 95 +++++-- .../__tests__/WorkerOrchestrator.test.ts | 164 +++++++++++- .../interfaces/IWorkerOrchestrator.ts | 3 + .../adapters/split/PDFSplitter.ts | 95 ++++--- .../__tests__/PDFSplitter.integration.test.ts | 75 ++++++ .../split/__tests__/PDFSplitter.test.ts | 251 +++++++++++++++++- .../infrastructure/config/worker.config.ts | 4 + .../migration.sql | 2 + src/core/infrastructure/db/schema.prisma | 1 + .../handlers/__tests__/task.handler.test.ts | 33 +++ src/main/ipc/handlers/task.handler.ts | 1 + 11 files changed, 658 insertions(+), 66 deletions(-) create mode 100644 src/core/infrastructure/db/migrations/20261105000000_add_task_split_interrupt_count/migration.sql diff --git a/src/core/application/services/WorkerOrchestrator.ts b/src/core/application/services/WorkerOrchestrator.ts index 1bd1e8d..03c8372 100644 --- a/src/core/application/services/WorkerOrchestrator.ts +++ b/src/core/application/services/WorkerOrchestrator.ts @@ -1,3 +1,4 @@ +import { promises as fs } from 'fs'; import { SplitterWorker, ConverterWorker, MergerWorker } from '../workers/index.js'; import { ImagePathUtil } from '../../infrastructure/adapters/split/index.js'; import fileLogic from '../../infrastructure/services/FileService.js'; @@ -183,7 +184,8 @@ export class WorkerOrchestrator implements IWorkerOrchestrator { * - Pages with status=PROCESSING and worker_id set (orphaned by crashed workers) * - Tasks with status=SPLITTING (orphaned splitter work) * - * These are reset to their previous state so new workers can pick them up. + * These are reset to their previous state so new workers can pick them up, + * except for splits interrupted too many times (see recoverInterruptedSplits). */ async cleanupOrphanedWork(): Promise { try { @@ -207,22 +209,9 @@ export class WorkerOrchestrator implements IWorkerOrchestrator { console.log(`[WorkerOrchestrator] Reset ${orphanedPages.count} orphaned pages to PENDING`); } - // Reset orphaned splitting tasks (SPLITTING -> PENDING) - // These are tasks that were being split when the app was closed - const orphanedSplittingTasks = await prisma.task.updateMany({ - where: { - status: TaskStatus.SPLITTING, - worker_id: { not: null }, - }, - data: { - status: TaskStatus.PENDING, - worker_id: null, - }, - }); - - if (orphanedSplittingTasks.count > 0) { - console.log(`[WorkerOrchestrator] Reset ${orphanedSplittingTasks.count} orphaned SPLITTING tasks to PENDING`); - } + // Recover tasks that were being split when the app was closed + const { requeued: orphanedSplittingTasks, failed: failedSplittingTasks } = + await this.recoverInterruptedSplits(); // Reset orphaned merging tasks (MERGING -> READY_TO_MERGE) // These are tasks that were being merged when the app was closed @@ -284,10 +273,11 @@ export class WorkerOrchestrator implements IWorkerOrchestrator { const result: CleanupResult = { orphanedPages: orphanedPages.count, - orphanedSplittingTasks: orphanedSplittingTasks.count, + orphanedSplittingTasks, + failedSplittingTasks, orphanedMergingTasks: orphanedMergingTasks.count, orphanedPendingPages, - total: orphanedPages.count + orphanedSplittingTasks.count + orphanedMergingTasks.count + orphanedPendingPages, + total: orphanedPages.count + orphanedSplittingTasks + failedSplittingTasks + orphanedMergingTasks.count + orphanedPendingPages, }; if (result.total === 0) { @@ -303,12 +293,79 @@ export class WorkerOrchestrator implements IWorkerOrchestrator { return { orphanedPages: 0, orphanedSplittingTasks: 0, + failedSplittingTasks: 0, orphanedMergingTasks: 0, orphanedPendingPages: 0, total: 0, }; } } + + /** + * Recover tasks left in SPLITTING by an abnormal shutdown. + * + * A split that brought the app down (e.g. out of memory) would do so again on every + * restart, so each interruption is counted and the task fails once the limit is reached. + * The database can't tell a crash from quitting mid-split, so both count. A false + * positive costs the user one click on retry, which starts the count over. + */ + private async recoverInterruptedSplits(): Promise<{ requeued: number; failed: number }> { + const interrupted = { + status: TaskStatus.SPLITTING, + worker_id: { not: null }, + }; + const maxInterruptedSplits = WORKER_CONFIG.splitter.maxInterruptedSplits; + + const exhaustedTasks = await prisma.task.findMany({ + where: { + ...interrupted, + split_interrupt_count: { gte: maxInterruptedSplits - 1 }, + }, + select: { id: true }, + }); + + if (exhaustedTasks.length > 0) { + const exhaustedIds = exhaustedTasks.map((t) => t.id); + await prisma.task.updateMany({ + where: { id: { in: exhaustedIds } }, + data: { + status: TaskStatus.FAILED, + worker_id: null, + split_interrupt_count: { increment: 1 }, + error: `Splitting was interrupted ${maxInterruptedSplits} times, possibly because the app crashed or ran out of memory. Retry the task, or split the PDF into smaller files.`, + }, + }); + await Promise.all(exhaustedIds.map((id) => this.removeSplitImages(id))); + console.warn(`[WorkerOrchestrator] Failed ${exhaustedIds.length} tasks interrupted ${maxInterruptedSplits} times while splitting`); + } + + const requeued = await prisma.task.updateMany({ + where: interrupted, + data: { + status: TaskStatus.PENDING, + worker_id: null, + split_interrupt_count: { increment: 1 }, + }, + }); + + if (requeued.count > 0) { + console.log(`[WorkerOrchestrator] Reset ${requeued.count} orphaned SPLITTING tasks to PENDING`); + } + + return { requeued: requeued.count, failed: exhaustedTasks.length }; + } + + /** + * Remove the pages a failed split left behind. Best effort: a leftover only takes + * disk space until the task is deleted, and must not stop the workers from starting. + */ + private async removeSplitImages(taskId: string): Promise { + try { + await fs.rm(fileLogic.getSplitDir(taskId), { recursive: true, force: true }); + } catch (error) { + console.warn(`[WorkerOrchestrator] Failed to remove split images of task ${taskId}:`, error); + } + } } // Create singleton instance for backward compatibility diff --git a/src/core/application/services/__tests__/WorkerOrchestrator.test.ts b/src/core/application/services/__tests__/WorkerOrchestrator.test.ts index 96b510d..3662a40 100644 --- a/src/core/application/services/__tests__/WorkerOrchestrator.test.ts +++ b/src/core/application/services/__tests__/WorkerOrchestrator.test.ts @@ -48,12 +48,22 @@ vi.mock('../../../infrastructure/adapters/split/index.js', () => ({ vi.mock('../../../infrastructure/services/FileService.js', () => ({ default: { - getUploadDir: vi.fn().mockReturnValue('/mock/uploads') + getUploadDir: vi.fn().mockReturnValue('/mock/uploads'), + getSplitDir: vi.fn((taskId: string) => `/mock/uploads/${taskId}/split`) + } +})) + +vi.mock('fs', () => ({ + promises: { + rm: vi.fn().mockResolvedValue(undefined) } })) vi.mock('../../../infrastructure/config/worker.config.js', () => ({ WORKER_CONFIG: { + splitter: { + maxInterruptedSplits: 3 + }, converter: { count: 3 } @@ -64,6 +74,7 @@ import { WorkerOrchestrator } from '../WorkerOrchestrator.js' import { SplitterWorker, ConverterWorker, MergerWorker } from '../../workers/index.js' import { ImagePathUtil } from '../../../infrastructure/adapters/split/index.js' import { prisma } from '../../../infrastructure/db/index.js' +import { promises as fs } from 'fs' // Cast to mock type for type safety const prismaMock = prisma as unknown as DeepMockProxy @@ -293,10 +304,148 @@ describe('WorkerOrchestrator', () => { }, data: { status: TaskStatus.PENDING, - worker_id: null + worker_id: null, + split_interrupt_count: { increment: 1 } } }) expect(result.orphanedSplittingTasks).toBe(2) + expect(result.failedSplittingTasks).toBe(0) + }) + + it('should fail SPLITTING tasks on their last allowed interruption', async () => { + vi.spyOn(console, 'warn').mockImplementation(() => {}) + prismaMock.task.findMany.mockResolvedValueOnce([{ id: 'crashing-task' }] as any) + prismaMock.task.updateMany.mockResolvedValue({ count: 0 } as any) + + const result = await orchestrator.cleanupOrphanedWork() + + expect(prismaMock.task.findMany).toHaveBeenCalledWith({ + where: { + status: TaskStatus.SPLITTING, + worker_id: { not: null }, + split_interrupt_count: { gte: 2 } + }, + select: { id: true } + }) + expect(prismaMock.task.updateMany).toHaveBeenCalledWith({ + where: { id: { in: ['crashing-task'] } }, + data: { + status: TaskStatus.FAILED, + worker_id: null, + split_interrupt_count: { increment: 1 }, + error: expect.stringContaining('interrupted 3 times') + } + }) + expect(result.failedSplittingTasks).toBe(1) + expect(result.total).toBe(1) + }) + + it('should remove the pages left behind by a task it fails', async () => { + vi.spyOn(console, 'warn').mockImplementation(() => {}) + prismaMock.task.findMany.mockResolvedValueOnce([{ id: 'crashing-task' }] as any) + + await orchestrator.cleanupOrphanedWork() + + expect(fs.rm).toHaveBeenCalledTimes(1) + expect(fs.rm).toHaveBeenCalledWith('/mock/uploads/crashing-task/split', { + recursive: true, + force: true + }) + }) + + it('should not remove any pages of tasks that are re-queued', async () => { + prismaMock.task.updateMany.mockResolvedValueOnce({ count: 2 } as any) + + await orchestrator.cleanupOrphanedWork() + + expect(fs.rm).not.toHaveBeenCalled() + }) + + it('should still recover other work when removing leftover pages fails', async () => { + const warnSpy = vi.spyOn(console, 'warn').mockImplementation(() => {}) + vi.mocked(fs.rm).mockRejectedValueOnce(new Error('EBUSY')) + prismaMock.task.findMany.mockResolvedValueOnce([{ id: 'crashing-task' }] as any) + prismaMock.task.updateMany + .mockResolvedValueOnce({ count: 1 } as any) // fail the exhausted task + .mockResolvedValueOnce({ count: 0 } as any) // SPLITTING -> PENDING + .mockResolvedValueOnce({ count: 2 } as any) // MERGING -> READY_TO_MERGE + + const result = await orchestrator.cleanupOrphanedWork() + + expect(result.failedSplittingTasks).toBe(1) + expect(result.orphanedMergingTasks).toBe(2) + expect(warnSpy).toHaveBeenCalledWith( + expect.stringContaining('crashing-task'), + expect.any(Error) + ) + }) + + describe('across restarts', () => { + type StoredTask = { id: string; status: number; worker_id: string | null; split_interrupt_count: number } + + /** Applies the task queries of the recovery to a task, as the database would. */ + function backTasksWith(task: StoredTask) { + const isInterrupted = () => task.status === TaskStatus.SPLITTING && task.worker_id !== null + const applyUpdate = (data: any) => { + task.status = data.status + task.worker_id = data.worker_id + task.split_interrupt_count += data.split_interrupt_count.increment + } + + prismaMock.task.findMany.mockImplementation((async ({ where }: any) => + where.split_interrupt_count && + isInterrupted() && + task.split_interrupt_count >= where.split_interrupt_count.gte + ? [{ id: task.id }] + : []) as any) + + prismaMock.task.updateMany.mockImplementation((async ({ where, data }: any) => { + const matches = where.id + ? where.id.in.includes(task.id) + : where.status === TaskStatus.SPLITTING && isInterrupted() + if (!matches) { + return { count: 0 } + } + applyUpdate(data) + return { count: 1 } + }) as any) + } + + /** The next start claims the task if it is queued, and it brings the app down again. */ + function crashWhileSplitting(task: StoredTask) { + if (task.status === TaskStatus.PENDING) { + task.status = TaskStatus.SPLITTING + task.worker_id = 'worker-of-a-dead-process' + } + } + + it('should stop re-queuing a task that crashes the app on every start', async () => { + vi.spyOn(console, 'warn').mockImplementation(() => {}) + const task: StoredTask = { + id: 'crashing-task', + status: TaskStatus.PENDING, + worker_id: null, + split_interrupt_count: 0 + } + backTasksWith(task) + + const statusAfterRestart: number[] = [] + for (let restart = 0; restart < 5; restart++) { + crashWhileSplitting(task) + await orchestrator.cleanupOrphanedWork() + statusAfterRestart.push(task.status) + } + + expect(statusAfterRestart).toEqual([ + TaskStatus.PENDING, + TaskStatus.PENDING, + TaskStatus.FAILED, + TaskStatus.FAILED, + TaskStatus.FAILED + ]) + expect(task.split_interrupt_count).toBe(3) + expect(task.worker_id).toBeNull() + }) }) it('should reset MERGING tasks to READY_TO_MERGE', async () => { @@ -326,10 +475,12 @@ describe('WorkerOrchestrator', () => { .mockResolvedValueOnce({ count: 0 } as any) // PROCESSING -> PENDING (first call) .mockResolvedValueOnce({ count: 4 } as any) // orphaned PENDING -> FAILED (second call) prismaMock.task.updateMany.mockResolvedValue({ count: 0 } as any) - prismaMock.task.findMany.mockResolvedValue([ - { id: 'failed-task-1' }, - { id: 'cancelled-task-2' }, - ] as any) + prismaMock.task.findMany + .mockResolvedValueOnce([] as any) // no SPLITTING task out of attempts + .mockResolvedValueOnce([ + { id: 'failed-task-1' }, + { id: 'cancelled-task-2' }, + ] as any) const result = await orchestrator.cleanupOrphanedWork() @@ -380,6 +531,7 @@ describe('WorkerOrchestrator', () => { expect(result).toEqual({ orphanedPages: 0, orphanedSplittingTasks: 0, + failedSplittingTasks: 0, orphanedMergingTasks: 0, orphanedPendingPages: 0, total: 0 diff --git a/src/core/application/services/interfaces/IWorkerOrchestrator.ts b/src/core/application/services/interfaces/IWorkerOrchestrator.ts index e1ee5a8..77b6928 100644 --- a/src/core/application/services/interfaces/IWorkerOrchestrator.ts +++ b/src/core/application/services/interfaces/IWorkerOrchestrator.ts @@ -24,7 +24,10 @@ export interface WorkerStatus { */ export interface CleanupResult { orphanedPages: number; + /** Interrupted splits put back in the queue */ orphanedSplittingTasks: number; + /** Interrupted splits failed instead, having been interrupted too many times */ + failedSplittingTasks: number; orphanedMergingTasks: number; orphanedPendingPages: number; total: number; diff --git a/src/core/infrastructure/adapters/split/PDFSplitter.ts b/src/core/infrastructure/adapters/split/PDFSplitter.ts index d8f4497..b8d6ea1 100644 --- a/src/core/infrastructure/adapters/split/PDFSplitter.ts +++ b/src/core/infrastructure/adapters/split/PDFSplitter.ts @@ -1,4 +1,4 @@ -import { pdfToPng } from 'pdf-to-png-converter'; +import { pdfToPng, type PdfToPngOptions } from 'pdf-to-png-converter'; import { PDFDocument } from 'pdf-lib'; import { promises as fs } from 'fs'; import path from 'path'; @@ -32,6 +32,8 @@ function isRetryablePdfError(error: Error): boolean { * - Sequential page numbering (page-1.png, page-2.png, ...) * - Output location independent of process.cwd() (pages are written by this class, * see renderPagesWithRetry) + * - Memory bounded by the batch size, not the page count: pages are rendered and + * written in batches, because pdfToPng holds every page it renders until it returns */ export class PDFSplitter implements ISplitter { private readonly uploadsDir: string; @@ -63,24 +65,40 @@ export class PDFSplitter implements ISplitter { // Retry with short delays to handle antivirus scanning or filesystem sync delays await FileWaitUtil.waitForFile(sourcePath, this.uploadsDir, taskId, filename, 'PDFSplitter'); - let renderedPages: RenderedPage[]; - try { - // Step 1: Get total page count with retry + // Step 1: Resolve the pages to render (defaults to all pages if no range is specified) + const pageNumbers = await this.withSourceErrors(sourcePath, filename, async () => { const totalPages = await this.getPDFPageCountWithRetry(sourcePath); + return PageRangeParser.parse(task.page_range, totalPages); + }); + + // Step 2: Render and write batch by batch. Only `pages` (metadata) outlives an iteration, + // so the PNG buffers of a batch can be collected before the next one is rendered. + const hasPageRange = Boolean(task.page_range && task.page_range.trim() !== ''); + const batchSize = WORKER_CONFIG.splitter.renderBatchSize; + const pages: PageInfo[] = []; + for (let start = 0; ; start += batchSize) { + // Without a range, pdf.js decides where the document ends, not the pdf-lib count: pdf-lib + // misses pages whose tree nodes lack a /Type, and those pages must not be dropped. + const batch = hasPageRange + ? pageNumbers.slice(start, start + batchSize) + : Array.from({ length: batchSize }, (_, i) => start + i + 1); + if (batch.length === 0) { + break; + } - // Step 2: Parse page range (defaults to all pages if not specified) - const pageNumbers = PageRangeParser.parse(task.page_range, totalPages); + const renderedPages = await this.withSourceErrors(sourcePath, filename, () => + this.renderPagesWithRetry(sourcePath, batch, pages.length) + ); - // Step 3: Render specified pages with retry - // Pass the page_range string to determine if user explicitly specified pages - renderedPages = await this.renderPagesWithRetry(sourcePath, pageNumbers, task.page_range); - } catch (error) { - throw await this.wrapError(error, sourcePath, filename); - } + // Not covered by withSourceErrors, so that an output failure is never reported + // as a problem with the source PDF. + pages.push(...(await this.writePagesWithRetry(taskId, renderedPages))); - // Step 4: Write the images. Kept apart from the steps above so that an output - // failure is never reported as a problem with the source PDF. - const pages = await this.writePagesWithRetry(taskId, renderedPages); + // Pages are ascending, so a short batch means the document has ended. + if (renderedPages.length < batch.length) { + break; + } + } return { pages, @@ -88,6 +106,21 @@ export class PDFSplitter implements ISplitter { }; } + /** + * Run a step that reads the source PDF, reporting its failure with a friendly message. + */ + private async withSourceErrors( + sourcePath: string, + filename: string, + step: () => Promise + ): Promise { + try { + return await step(); + } catch (error) { + throw await this.wrapError(error, sourcePath, filename); + } + } + /** * Get PDF page count with retry logic. * Uses pdf-lib for accurate page count without converting pages. @@ -114,47 +147,45 @@ export class PDFSplitter implements ISplitter { } /** - * Render specified pages to in-memory PNGs with retry logic. + * Render one batch of pages to in-memory PNGs with retry logic. + * + * @param pdfPath - Source PDF + * @param sourcePages - Source page numbers of this batch, ascending + * @param pagesBefore - Pages produced by the previous batches, which the numbering continues from */ private async renderPagesWithRetry( pdfPath: string, - pageNumbers: number[], - pageRange?: string | null + sourcePages: number[], + pagesBefore: number ): Promise { return this.withRetry(async () => { // IMPORTANT: do not pass outputFolder to pdf-to-png-converter. It resolves it with // path.join(process.cwd(), outputFolder), which cannot address a directory on another // Windows drive or UNC share (#75). We take the PNG buffers and write them ourselves. - const options: any = { + const options: PdfToPngOptions = { returnPageContent: true, viewportScale: WORKER_CONFIG.splitter.viewportScale, - strictPagesToProcess: false, + pagesToProcess: sourcePages, verbosityLevel: 0, }; - // Only add pagesToProcess if user explicitly specified a page range - // If pageRange is empty/null, let pdf-to-png-converter process all pages - const hasPageRange = Boolean(pageRange && pageRange.trim() !== ''); - if (hasPageRange) { - options.pagesToProcess = pageNumbers; // Array of page numbers - } - // Otherwise, don't specify pagesToProcess to convert all pages - const result = await pdfToPng(pdfPath, options); - if (!result || result.length === 0) { + // pdfToPng skips pages beyond the end of the document, so a later batch may come back + // short or empty, but the document as a whole must produce something. + if (!result || (result.length === 0 && pagesBefore === 0)) { throw new Error('PDF conversion produced no output'); } return result.map((output, i) => { - const page = i + 1; // Sequential numbering + const page = pagesBefore + i + 1; // Sequential numbering if (!output.content) { throw new Error(`PDF conversion produced no image for page ${page}`); } return { page, - // If page range was specified, use the mapped page numbers; otherwise use sequential numbering - pageSource: hasPageRange ? pageNumbers[i] : page, + // Only trailing pages can be skipped, so the indexes of `sourcePages` still line up. + pageSource: sourcePages[i], content: output.content, }; }); diff --git a/src/core/infrastructure/adapters/split/__tests__/PDFSplitter.integration.test.ts b/src/core/infrastructure/adapters/split/__tests__/PDFSplitter.integration.test.ts index 3237028..019c74d 100644 --- a/src/core/infrastructure/adapters/split/__tests__/PDFSplitter.integration.test.ts +++ b/src/core/infrastructure/adapters/split/__tests__/PDFSplitter.integration.test.ts @@ -5,6 +5,7 @@ import path from 'path'; import { PDFDocument } from 'pdf-lib'; import { PDFSplitter } from '../PDFSplitter.js'; import { ImagePathUtil } from '../ImagePathUtil.js'; +import { WORKER_CONFIG } from '../../../config/worker.config.js'; /** * Runs against the real pdf-to-png-converter, pdf-lib and file system, because @@ -70,6 +71,80 @@ describe('PDFSplitter with the real converter', () => { expect(await fs.readdir(splitDir)).toEqual(['page-1.png']); }); + describe('across render batches', () => { + const BATCH_SIZE = WORKER_CONFIG.splitter.renderBatchSize; + const PAGE_COUNT = BATCH_SIZE * 2 + 3; + + /** Every page gets its own width, which identifies it in the rendered PNG. */ + const widthOf = (sourcePage: number) => PAGE_SIZE_PT + sourcePage; + + beforeEach(async () => { + const pdf = await PDFDocument.create(); + for (let page = 1; page <= PAGE_COUNT; page++) { + pdf.addPage([widthOf(page), PAGE_SIZE_PT]); + } + await fs.writeFile(path.join(uploadsDir, TASK_ID, 'doc.pdf'), await pdf.save()); + }); + + async function expectPages(pageRange: string, sourcePages: number[]) { + const result = await splitter.split({ id: TASK_ID, filename: 'doc.pdf', page_range: pageRange }); + + expect(result.totalPages).toBe(sourcePages.length); + expect(result.pages).toEqual( + sourcePages.map((pageSource, i) => ({ + page: i + 1, + pageSource, + imagePath: path.join(splitDir, `page-${i + 1}.png`), + })) + ); + expect((await fs.readdir(splitDir)).sort()).toEqual( + sourcePages.map((_, i) => `page-${i + 1}.png`).sort() + ); + for (const [i, sourcePage] of sourcePages.entries()) { + const png = await fs.readFile(path.join(splitDir, `page-${i + 1}.png`)); + expect(png.readUInt32BE(16)).toBe(widthOf(sourcePage) * 2); + } + } + + it('should render every page in order', async () => { + await expectPages('', Array.from({ length: PAGE_COUNT }, (_, i) => i + 1)); + }); + + it('should render a page range in order', async () => { + const lastPages = [PAGE_COUNT - 1, PAGE_COUNT]; + await expectPages( + `1,3-${BATCH_SIZE + 4},${lastPages.join(',')}`, + [1, ...Array.from({ length: BATCH_SIZE + 2 }, (_, i) => i + 3), ...lastPages] + ); + }); + }); + + it('should keep a page that pdf-lib does not count because its node has no /Type', async () => { + const objects = [ + '<< /Type /Catalog /Pages 2 0 R >>', + '<< /Type /Pages /Count 3 /Kids [3 0 R 4 0 R 5 0 R] >>', + '<< /Type /Page /Parent 2 0 R /MediaBox [0 0 100 100] >>', + '<< /Parent 2 0 R /MediaBox [0 0 100 100] >>', + '<< /Type /Page /Parent 2 0 R /MediaBox [0 0 100 100] >>', + ]; + let pdf = '%PDF-1.4\n'; + const offsets = objects.map((object, i) => { + const offset = pdf.length; + pdf += `${i + 1} 0 obj\n${object}\nendobj\n`; + return offset; + }); + const xref = pdf.length; + pdf += `xref\n0 ${objects.length + 1}\n0000000000 65535 f \n`; + pdf += offsets.map((offset) => `${String(offset).padStart(10, '0')} 00000 n \n`).join(''); + pdf += `trailer\n<< /Size ${objects.length + 1} /Root 1 0 R >>\nstartxref\n${xref}\n%%EOF\n`; + await fs.writeFile(path.join(uploadsDir, TASK_ID, 'doc.pdf'), pdf, 'latin1'); + + const result = await splitter.split({ id: TASK_ID, filename: 'doc.pdf', page_range: '' }); + + expect(result.pages.map((page) => page.pageSource)).toEqual([1, 2, 3]); + expect((await fs.readdir(splitDir)).sort()).toEqual(['page-1.png', 'page-2.png', 'page-3.png']); + }); + it('should not blame the source PDF when the split dir cannot be created', async () => { vi.spyOn(splitter as any, 'sleep').mockResolvedValue(undefined); vi.spyOn(console, 'warn').mockImplementation(() => {}); diff --git a/src/core/infrastructure/adapters/split/__tests__/PDFSplitter.test.ts b/src/core/infrastructure/adapters/split/__tests__/PDFSplitter.test.ts index 71955b5..b13a5fd 100644 --- a/src/core/infrastructure/adapters/split/__tests__/PDFSplitter.test.ts +++ b/src/core/infrastructure/adapters/split/__tests__/PDFSplitter.test.ts @@ -40,6 +40,22 @@ vi.mock('fs', () => ({ import { pdfToPng } from 'pdf-to-png-converter'; import { PDFDocument } from 'pdf-lib'; import { FileWaitUtil } from '../FileWaitUtil.js'; +import { WORKER_CONFIG } from '../../../config/worker.config.js'; + +/** Renders the requested pages the way pdf-to-png-converter does, stopping at the document's end. */ +function fakePdfToPng(pagesInDocument: number) { + return async (_pdf: unknown, options?: { pagesToProcess?: number[] }) => + (options?.pagesToProcess ?? Array.from({ length: pagesInDocument }, (_, i) => i + 1)) + .filter((page) => page <= pagesInDocument) + .map((page) => ({ + path: '', + name: `doc_page_${page}.png`, + pageNumber: page, + content: Buffer.from(`source-${page}`), + width: 100, + height: 100, + })); +} describe('PDFSplitter', () => { const uploadsDir = '/mock/uploads'; @@ -466,21 +482,238 @@ describe('PDFSplitter', () => { getPageCount: () => 200, } as any); - // Mock 100 pages - const pages = Array.from({ length: 100 }, (_, i) => ({ - path: `/tmp/page-${i + 1}.png`, - name: `page-${i + 1}.png`, - page: i + 1, - content: Buffer.from(''), - })); - - vi.mocked(pdfToPng).mockResolvedValueOnce(pages); + vi.mocked(pdfToPng).mockImplementation(fakePdfToPng(200) as any); vi.mocked(fs.mkdir).mockResolvedValue(undefined); const result = await splitter.split(task); expect(result.totalPages).toBe(100); expect(result.pages).toHaveLength(100); + vi.mocked(pdfToPng).mockReset(); + }); + }); + + describe('batched rendering', () => { + const BATCH_SIZE = WORKER_CONFIG.splitter.renderBatchSize; + const taskId = 'task-batch'; + const task = { id: taskId, filename: 'document.pdf', page_range: '' }; + const pagePath = (page: number) => path.join(uploadsDir, taskId, 'split', `page-${page}.png`); + const range = (from: number, to: number) => + Array.from({ length: to - from + 1 }, (_, i) => from + i); + const renderedBatches = () => + vi.mocked(pdfToPng).mock.calls.map(([, options]) => options?.pagesToProcess); + + function givenDocument(pagesByPdfLib: number, pagesByPdfJs: number = pagesByPdfLib) { + vi.mocked(PDFDocument.load).mockResolvedValue({ getPageCount: () => pagesByPdfLib } as any); + vi.mocked(pdfToPng).mockImplementation(fakePdfToPng(pagesByPdfJs) as any); + } + + beforeEach(() => { + vi.spyOn(splitter as any, 'sleep').mockResolvedValue(undefined); + vi.spyOn(console, 'warn').mockImplementation(() => {}); + }); + + afterEach(() => { + vi.mocked(pdfToPng).mockReset(); + vi.mocked(fs.writeFile).mockReset(); + vi.restoreAllMocks(); + }); + + it('should render all pages in batches of the configured size', async () => { + const pageCount = BATCH_SIZE * 2 + 3; + givenDocument(pageCount); + + const result = await splitter.split(task); + + expect(renderedBatches()).toEqual([ + range(1, BATCH_SIZE), + range(BATCH_SIZE + 1, BATCH_SIZE * 2), + range(BATCH_SIZE * 2 + 1, BATCH_SIZE * 3), + ]); + expect(result.totalPages).toBe(pageCount); + expect(result.pages).toEqual( + range(1, pageCount).map((page) => ({ page, pageSource: page, imagePath: pagePath(page) })) + ); + }); + + it('should find the end of the document with an empty batch when the pages fit exactly', async () => { + givenDocument(BATCH_SIZE * 2); + + const result = await splitter.split(task); + + expect(pdfToPng).toHaveBeenCalledTimes(3); + expect(result.totalPages).toBe(BATCH_SIZE * 2); + }); + + it('should not render a further batch when a range is used up exactly', async () => { + givenDocument(BATCH_SIZE * 3); + + const result = await splitter.split({ ...task, page_range: `1-${BATCH_SIZE * 2}` }); + + expect(pdfToPng).toHaveBeenCalledTimes(2); + expect(result.totalPages).toBe(BATCH_SIZE * 2); + }); + + it('should render the pages pdf.js finds when pdf-lib counts fewer', async () => { + const pagesByPdfJs = BATCH_SIZE + 3; + givenDocument(2, pagesByPdfJs); + + const result = await splitter.split(task); + + expect(result.pages.map((page) => page.pageSource)).toEqual(range(1, pagesByPdfJs)); + }); + + it('should stop a range at the end of the document as pdf.js counts it', async () => { + const pagesByPdfJs = BATCH_SIZE + 2; + givenDocument(BATCH_SIZE * 5, pagesByPdfJs); + + const result = await splitter.split({ ...task, page_range: `1-${BATCH_SIZE * 4}` }); + + expect(pdfToPng).toHaveBeenCalledTimes(2); + expect(result.pages.map((page) => page.pageSource)).toEqual(range(1, pagesByPdfJs)); + }); + + it('should write each page image with the content of its source page', async () => { + const pageCount = BATCH_SIZE + 2; + givenDocument(pageCount); + + await splitter.split(task); + + expect(vi.mocked(fs.writeFile).mock.calls).toEqual( + range(1, pageCount).map((page) => [pagePath(page), Buffer.from(`source-${page}`)]) + ); + }); + + it('should number pages sequentially and keep source pages for a range across batches', async () => { + givenDocument(100); + const sourcePages = [2, 3, ...range(10, 10 + BATCH_SIZE), 50, 99]; + + const result = await splitter.split({ ...task, page_range: `2-3,10-${10 + BATCH_SIZE},50,99` }); + + expect(renderedBatches().flat()).toEqual(sourcePages); + expect(result.pages).toEqual( + sourcePages.map((pageSource, i) => ({ + page: i + 1, + pageSource, + imagePath: pagePath(i + 1), + })) + ); + expect(fs.writeFile).toHaveBeenCalledWith(pagePath(3), Buffer.from(`source-${sourcePages[2]}`)); + }); + + it('should never hold more than one batch of rendered pages', async () => { + const pageCount = BATCH_SIZE * 5 + 1; + givenDocument(pageCount); + const render = fakePdfToPng(pageCount); + let unwritten = 0; + let mostUnwritten = 0; + vi.mocked(pdfToPng).mockImplementation((async (pdf: unknown, options: any) => { + const pages = await render(pdf, options); + unwritten += pages.length; + mostUnwritten = Math.max(mostUnwritten, unwritten); + return pages; + }) as any); + vi.mocked(fs.writeFile).mockImplementation((async () => { + unwritten--; + }) as any); + + await splitter.split(task); + + expect(unwritten).toBe(0); + expect(mostUnwritten).toBe(BATCH_SIZE); + }); + + it('should not keep page images in the result', async () => { + givenDocument(BATCH_SIZE + 1); + + const result = await splitter.split(task); + + result.pages.forEach((page) => expect(page).not.toHaveProperty('content')); + }); + + it('should retry only the batch whose rendering failed', async () => { + const pageCount = BATCH_SIZE * 2 - 1; + givenDocument(pageCount); + const render = fakePdfToPng(pageCount); + vi.mocked(pdfToPng) + .mockImplementationOnce(render as any) + .mockRejectedValueOnce(new Error('Transient render error')); + + const result = await splitter.split(task); + + expect(result.totalPages).toBe(pageCount); + expect(renderedBatches()).toEqual([ + range(1, BATCH_SIZE), + range(BATCH_SIZE + 1, BATCH_SIZE * 2), + range(BATCH_SIZE + 1, BATCH_SIZE * 2), + ]); + expect(fs.writeFile).toHaveBeenCalledTimes(pageCount); + }); + + it('should retry a failed write of a later batch without rendering again', async () => { + givenDocument(BATCH_SIZE + 1); + vi.mocked(fs.writeFile).mockImplementation((async (target: string) => { + if (target === pagePath(BATCH_SIZE + 1) && vi.mocked(fs.writeFile).mock.calls.length === BATCH_SIZE + 1) { + throw new Error('EBUSY: resource busy'); + } + }) as any); + + const result = await splitter.split(task); + + expect(result.totalPages).toBe(BATCH_SIZE + 1); + expect(pdfToPng).toHaveBeenCalledTimes(2); + expect(fs.writeFile).toHaveBeenCalledTimes(BATCH_SIZE + 2); + }); + + it('should report a failure in a later batch as a source problem and stop', async () => { + givenDocument(BATCH_SIZE * 3); + const render = fakePdfToPng(BATCH_SIZE * 3); + vi.mocked(pdfToPng).mockImplementation((async (pdf: unknown, options: any) => { + if (options.pagesToProcess[0] > BATCH_SIZE) { + throw new Error('Render exploded'); + } + return render(pdf, options); + }) as any); + + const failure = splitter.split(task); + + await expect(failure).rejects.toThrow('Failed to split PDF document.pdf'); + await expect(failure).rejects.toThrow('Render exploded'); + // The third batch is never attempted, and only the first one reached the disk. + expect(renderedBatches().every((batch) => batch![0] <= BATCH_SIZE + 1)).toBe(true); + expect(fs.writeFile).toHaveBeenCalledTimes(BATCH_SIZE); + }); + + it('should not retry a later batch for a non-retryable error', async () => { + givenDocument(BATCH_SIZE + 1); + const render = fakePdfToPng(BATCH_SIZE + 1); + vi.mocked(pdfToPng) + .mockImplementationOnce(render as any) + .mockRejectedValueOnce(new Error('PDF is encrypted')); + + await expect(splitter.split(task)).rejects.toThrow(/Cannot process password-protected PDF/); + + expect(pdfToPng).toHaveBeenCalledTimes(2); + }); + + it.each([ + { scenario: 'partway through a batch', pagesByPdfJs: BATCH_SIZE + 1 }, + { scenario: 'exactly at a batch boundary', pagesByPdfJs: BATCH_SIZE }, + ])('should keep the rendered pages when pdf.js ends the document $scenario', async ({ pagesByPdfJs }) => { + givenDocument(BATCH_SIZE + 5, pagesByPdfJs); + + const result = await splitter.split(task); + + expect(result.totalPages).toBe(pagesByPdfJs); + expect(result.pages.map((page) => page.pageSource)).toEqual(range(1, pagesByPdfJs)); + }); + + it('should fail when the document renders nothing at all', async () => { + givenDocument(BATCH_SIZE + 1, 0); + + await expect(splitter.split(task)).rejects.toThrow('PDF conversion produced no output'); + + expect(fs.writeFile).not.toHaveBeenCalled(); }); }); }); diff --git a/src/core/infrastructure/config/worker.config.ts b/src/core/infrastructure/config/worker.config.ts index 05d1f7e..62317af 100644 --- a/src/core/infrastructure/config/worker.config.ts +++ b/src/core/infrastructure/config/worker.config.ts @@ -16,6 +16,10 @@ export const WORKER_CONFIG = { viewportScale: 2.0, /** Output image format */ imageFormat: 'png' as const, + /** Pages rendered per batch. PNGs of a batch stay in memory until it is written to disk */ + renderBatchSize: 10, + /** Interrupted splits (crash, forced quit) of a task before it fails instead of being re-queued */ + maxInterruptedSplits: 3, /** Maximum retry attempts for transient errors */ maxRetries: 3, /** Base retry delay in milliseconds (exponential backoff) */ diff --git a/src/core/infrastructure/db/migrations/20261105000000_add_task_split_interrupt_count/migration.sql b/src/core/infrastructure/db/migrations/20261105000000_add_task_split_interrupt_count/migration.sql new file mode 100644 index 0000000..89a54a0 --- /dev/null +++ b/src/core/infrastructure/db/migrations/20261105000000_add_task_split_interrupt_count/migration.sql @@ -0,0 +1,2 @@ +-- AlterTable +ALTER TABLE "Task" ADD COLUMN "split_interrupt_count" INTEGER NOT NULL DEFAULT 0; diff --git a/src/core/infrastructure/db/schema.prisma b/src/core/infrastructure/db/schema.prisma index 77e5871..1022d55 100644 --- a/src/core/infrastructure/db/schema.prisma +++ b/src/core/infrastructure/db/schema.prisma @@ -53,6 +53,7 @@ model Task { // Worker 追踪 worker_id String? // 当前处理的 worker ID + split_interrupt_count Int @default(0) // 拆页被异常退出中断的次数，用于终止循环崩溃 // 结果和错误 merged_path String? // 合并后的文件路径 diff --git a/src/main/ipc/handlers/__tests__/task.handler.test.ts b/src/main/ipc/handlers/__tests__/task.handler.test.ts index 5b46a43..3ee07b3 100644 --- a/src/main/ipc/handlers/__tests__/task.handler.test.ts +++ b/src/main/ipc/handlers/__tests__/task.handler.test.ts @@ -349,6 +349,39 @@ describe('Task Handler', () => { ) }) + it('should give a retried task a fresh count of interrupted splits', async () => { + const taskUpdate = vi.fn().mockResolvedValue({ id: 'task-1', status: 1 }) + + mockPrisma.$transaction.mockImplementation(async (callback: any) => + callback({ + task: { + findUnique: vi.fn().mockResolvedValue({ + id: 'task-1', + status: 0, + provider: 1, + model: 'gpt-4o', + model_name: 'GPT-4o | OpenAI', + split_interrupt_count: 3, + }), + update: taskUpdate, + }, + taskDetail: { + count: vi.fn().mockResolvedValue(0), + updateMany: vi.fn(), + }, + }) + ) + + const handler = handlers.get('task:retry') + const result = await handler!({}, 'task-1') + + expect(result.success).toBe(true) + expect(taskUpdate).toHaveBeenCalledWith({ + where: { id: 'task-1' }, + data: expect.objectContaining({ status: 1, split_interrupt_count: 0 }), + }) + }) + it('should retry task with model override', async () => { const taskDetailUpdateMany = vi.fn().mockResolvedValue({ count: 3 }) const taskUpdate = vi.fn().mockResolvedValue({ diff --git a/src/main/ipc/handlers/task.handler.ts b/src/main/ipc/handlers/task.handler.ts index 5dce4a1..d0c2a41 100644 --- a/src/main/ipc/handlers/task.handler.ts +++ b/src/main/ipc/handlers/task.handler.ts @@ -237,6 +237,7 @@ export function registerTaskHandlers() { error: null, merged_path: null, worker_id: null, + split_interrupt_count: 0, }, }); }, {