diff --git a/src/main.ts b/src/main.ts index a3c643c..a4c9231 100644 --- a/src/main.ts +++ b/src/main.ts @@ -19,7 +19,7 @@ import { tBackend as tBackendCore, type BackendMessageKey } from './main/domain/ import { watchRendererChanges } from './main/dev-reload'; import { createPausableOutput, type PausableOutput } from './main/domain/pausable-output'; import { PartialDownloadRegistry } from './main/domain/partial-download'; -import { QueueProcessRegistry, QueueRunLifecycle } from './main/queue/process-registry'; +import { QueueProcessRegistry, QueueRunLifecycle, waitForChildProcessExit } from './main/queue/process-registry'; import type { DbHandle } from './main/infra/db'; import { normalizeLogin, @@ -3348,41 +3348,51 @@ async function concatVideoFiles(inputFiles: string[], outputFile: string, itemId outputFile ]; - return await new Promise((resolve) => { - const proc = spawn(ffmpeg, args, { windowsHide: true }); - const registration = itemId - ? queueProcessRegistry.register(itemId, 'post-processing', { - kill: () => proc.kill(), - wait: waitForChildProcessClose(proc), - cleanup: () => { - try { fs.rmSync(outputFile, { force: true }); } catch { } - try { fs.rmSync(listFile, { force: true }); } catch { } - }, - }) - : null; - let stderrBuf = ''; - proc.stderr?.on('data', (chunk: Buffer) => { stderrBuf += chunk.toString(); }); - proc.on('close', (code) => { - registration?.release(); - try { fs.unlinkSync(listFile); } catch { /* ignore */ } - if (code === 0 && (!itemId || !queueProcessRegistry.isCancelled(itemId)) && fs.existsSync(outputFile) && fs.statSync(outputFile).size > 0) { - appendDebugLog('concat-ok', { output: outputFile, parts: inputFiles.length }); - resolve(true); - } else { - appendDebugLog('concat-failed', { code, stderrTail: stderrBuf.slice(-400) }); - try { - if (fs.existsSync(outputFile)) fs.unlinkSync(outputFile); - } catch { /* ignore */ } - resolve(false); - } - }); - proc.on('error', (err) => { - registration?.release(); - try { fs.unlinkSync(listFile); } catch { /* ignore */ } - appendDebugLog('concat-spawn-error', String(err)); - resolve(false); - }); - }); + try { + while (true) { + const success = await new Promise((resolve) => { + const proc = spawn(ffmpeg, args, { windowsHide: true }); + const registration = itemId + ? queueProcessRegistry.register(itemId, 'post-processing', { + kill: () => proc.kill(), + wait: () => waitForChildProcessExit(proc), + pause: async () => { + try { proc.kill(); } catch { } + await waitForChildProcessExit(proc); + }, + cleanup: () => { + try { fs.rmSync(outputFile, { force: true }); } catch { } + try { fs.rmSync(listFile, { force: true }); } catch { } + }, + }) + : null; + let stderrBuf = ''; + proc.stderr?.on('data', (chunk: Buffer) => { stderrBuf += chunk.toString(); }); + proc.on('close', (code) => { + registration?.release(); + if (code === 0 && (!itemId || (!queueProcessRegistry.isCancelled(itemId) && !queueProcessRegistry.isPaused(itemId))) && fs.existsSync(outputFile) && fs.statSync(outputFile).size > 0) { + appendDebugLog('concat-ok', { output: outputFile, parts: inputFiles.length }); + resolve(true); + } else { + appendDebugLog('concat-failed', { code, stderrTail: stderrBuf.slice(-400) }); + try { fs.rmSync(outputFile, { force: true }); } catch { } + resolve(false); + } + }); + proc.on('error', (err) => { + registration?.release(); + appendDebugLog('concat-spawn-error', String(err)); + resolve(false); + }); + }); + if (success) return true; + if (!itemId || !queueProcessRegistry.isPaused(itemId)) return false; + await queueProcessRegistry.whenResumed(itemId); + if (queueProcessRegistry.isCancelled(itemId)) return false; + } + } finally { + try { fs.rmSync(listFile, { force: true }); } catch { } + } } async function cutVideo( @@ -3590,8 +3600,11 @@ async function mergeVideos( const registration = itemId ? queueProcessRegistry.register(itemId, 'merge', { kill: () => proc.kill(), - wait: waitForChildProcessClose(proc), - pause: () => proc.kill(), + wait: () => waitForChildProcessExit(proc), + pause: async () => { + try { proc.kill(); } catch { } + await waitForChildProcessExit(proc); + }, cleanup: () => { try { fs.rmSync(outputFile, { force: true }); } catch { } try { fs.rmSync(concatFile, { force: true }); } catch { } @@ -3707,8 +3720,11 @@ async function splitMergedFile( const registration = itemId ? queueProcessRegistry.register(itemId, 'split', { kill: () => proc.kill(), - wait: waitForChildProcessClose(proc), - pause: () => proc.kill(), + wait: () => waitForChildProcessExit(proc), + pause: async () => { + try { proc.kill(); } catch { } + await waitForChildProcessExit(proc); + }, cleanup: () => { try { fs.rmSync(outputFile, { force: true }); } catch { } }, }) : null; @@ -3806,7 +3822,7 @@ function downloadVODPart( const outputFinished = output.finished.then(() => null, (error) => error); const processRegistration = queueProcessRegistry.register(itemId, 'streamlink', { kill: () => proc.kill(), - wait: waitForChildProcessClose(proc), + wait: () => waitForChildProcessExit(proc), pause: () => output.pause(), resume: () => output.resume(), cancel: () => output.cancel(), @@ -8285,22 +8301,6 @@ let shutdownCleanupDone = false; let quitAfterCleanup = false; let shutdownPromise: Promise | null = null; -function waitForChildProcessClose(process: ChildProcess | null, timeoutMs = 5000): Promise { - if (!process || process.exitCode !== null || process.signalCode !== null) return Promise.resolve(); - return new Promise((resolve) => { - let settled = false; - const finish = (): void => { - if (settled) return; - settled = true; - clearTimeout(timer); - resolve(); - }; - const timer = setTimeout(finish, timeoutMs); - process.once('close', finish); - process.once('error', finish); - }); -} - async function shutdownCleanup(reason: 'window-all-closed' | 'before-quit'): Promise { if (shutdownCleanupDone) return; shutdownCleanupDone = true; @@ -8348,7 +8348,7 @@ async function shutdownCleanup(reason: 'window-all-closed' | 'before-quit'): Pro if (currentEditorProcess) { const editorProcess = currentEditorProcess; try { editorProcess.kill(); } catch { /* already exited */ } - await waitForChildProcessClose(editorProcess); + await waitForChildProcessExit(editorProcess); currentEditorProcess = null; } @@ -8357,7 +8357,7 @@ async function shutdownCleanup(reason: 'window-all-closed' | 'before-quit'): Pro for (const process of exportProcesses) { try { process.kill(); } catch { } } - await Promise.all(exportProcesses.map((process) => waitForChildProcessClose(process))); + await Promise.all(exportProcesses.map((process) => waitForChildProcessExit(process))); if (currentCutterProcess && exportProcesses.includes(currentCutterProcess)) currentCutterProcess = null; const mediaProcesses = [...currentCutterMediaProcesses, ...currentCutterWaveformProcesses, ...currentCutterProbeProcesses, ...currentCutterInfoProcesses, ...currentCutterPreviewProcesses]; cancelCutterMediaPreparation(); @@ -8367,7 +8367,7 @@ async function shutdownCleanup(reason: 'window-all-closed' | 'before-quit'): Pro for (const process of currentCutterInfoProcesses) { try { process.kill(); } catch { } } - await Promise.all(mediaProcesses.map((process) => waitForChildProcessClose(process))); + await Promise.all(mediaProcesses.map((process) => waitForChildProcessExit(process))); removeCutterPreviewDirectory(cutterMediaJob?.previewDirectory || null); if (currentCutterPartialFile) { try { fs.rmSync(currentCutterPartialFile, { force: true }); } catch { } diff --git a/src/main/queue/process-lifecycle.integration.test.ts b/src/main/queue/process-lifecycle.integration.test.ts new file mode 100644 index 0000000..33794a1 --- /dev/null +++ b/src/main/queue/process-lifecycle.integration.test.ts @@ -0,0 +1,96 @@ +import { spawn } from 'node:child_process'; +import { once } from 'node:events'; +import { mkdtempSync, rmSync, writeFileSync, existsSync } from 'node:fs'; +import { tmpdir } from 'node:os'; +import { join } from 'node:path'; +import { describe, expect, it } from 'vitest'; +import { QueueProcessRegistry, QueueRunLifecycle, waitForChildProcessExit } from './process-registry'; + +function waitForExit(process: ReturnType): Promise { + if (process.exitCode !== null || process.signalCode !== null) return Promise.resolve(); + return new Promise((resolve) => { + process.once('close', () => resolve()); + process.once('error', () => resolve()); + }); +} + +describe('queue process lifecycle integration', () => { + it('keeps quick resume behind a real child pause without deleting retry output', async () => { + const directory = mkdtempSync(join(tmpdir(), 'tvm-queue-pause-')); + const retryFile = join(directory, 'merge-retry.mp4'); + const child = spawn(process.execPath, ['-e', 'setInterval(() => undefined, 1000)'], { windowsHide: true }); + const registry = new QueueProcessRegistry(); + let resumedAfterExit = false; + + try { + writeFileSync(retryFile, 'retry'); + await once(child, 'spawn'); + registry.register('item-a', 'merge', { + kill: () => child.kill(), + wait: () => waitForChildProcessExit(child, 30), + pause: async () => { + child.kill(); + await waitForChildProcessExit(child, 30); + }, + resume: () => { + resumedAfterExit = child.exitCode !== null || child.signalCode !== null; + }, + cleanup: () => rmSync(retryFile, { force: true }), + }); + + const pausing = registry.pauseItem('item-a'); + const resuming = registry.resumeItem('item-a'); + expect(registry.isPaused('item-a')).toBe(true); + + await Promise.all([pausing, resuming]); + + expect(resumedAfterExit).toBe(true); + expect(registry.isPaused('item-a')).toBe(false); + expect(existsSync(retryFile)).toBe(true); + } finally { + if (child.exitCode === null && child.signalCode === null) child.kill(); + await waitForExit(child); + rmSync(directory, { recursive: true, force: true }); + } + }); + + it('waits for a late-cancelled real child before cleanup and final persistence', async () => { + const directory = mkdtempSync(join(tmpdir(), 'tvm-queue-lifecycle-')); + const partialFile = join(directory, 'output.mp4.partial'); + const child = spawn(process.execPath, ['-e', 'setInterval(() => undefined, 1000)'], { windowsHide: true }); + const registry = new QueueProcessRegistry(); + const lifecycle = new QueueRunLifecycle(registry); + const sequence: string[] = []; + + try { + writeFileSync(partialFile, 'partial'); + await once(child, 'spawn'); + const childExited = waitForExit(child); + lifecycle.schedule(async () => childExited); + registry.register('item-a', 'merge', { + kill: () => undefined, + wait: () => waitForChildProcessExit(child, 30), + cleanup: () => { + expect(child.exitCode !== null || child.signalCode !== null).toBe(true); + rmSync(partialFile, { force: true }); + sequence.push('cleanup'); + }, + }); + + await new Promise((resolve) => setTimeout(resolve, 50)); + expect(child.exitCode).toBeNull(); + expect(child.signalCode).toBeNull(); + expect(existsSync(partialFile)).toBe(true); + await lifecycle.shutdown(() => undefined, () => { + expect(existsSync(partialFile)).toBe(false); + sequence.push('persist'); + }); + + expect(sequence).toEqual(['cleanup', 'persist']); + } finally { + if (child.exitCode === null && child.signalCode === null) child.kill(); + await waitForExit(child); + rmSync(directory, { recursive: true, force: true }); + } + }); +}); diff --git a/src/main/queue/process-registry.test.ts b/src/main/queue/process-registry.test.ts index 2e42680..ed8c5a3 100644 --- a/src/main/queue/process-registry.test.ts +++ b/src/main/queue/process-registry.test.ts @@ -12,7 +12,7 @@ function deferred(): { promise: Promise; resolve: () => void } { function createResource(wait: Promise = Promise.resolve()): QueueProcessResource { return { kill: vi.fn(), - wait, + wait: () => wait, pause: vi.fn(), resume: vi.fn(), cancel: vi.fn(async () => undefined), @@ -70,6 +70,37 @@ describe('QueueProcessRegistry', () => { expect(registry.isPaused('item-a')).toBe(false); }); + it('keeps pause latched until existing and newly registered resources finish pausing', async () => { + const registry = new QueueProcessRegistry(); + const firstPaused = deferred(); + const latePaused = deferred(); + const first = createResource(); + const late = createResource(); + first.pause = vi.fn(() => firstPaused.promise); + late.pause = vi.fn(() => latePaused.promise); + + registry.register('item-a', 'merge', first); + const pausing = registry.pauseItem('item-a'); + registry.register('item-a', 'post-processing', late); + const resuming = registry.resumeItem('item-a'); + + await Promise.resolve(); + expect(registry.isPaused('item-a')).toBe(true); + expect(first.resume).not.toHaveBeenCalled(); + expect(late.pause).toHaveBeenCalledOnce(); + + firstPaused.resolve(); + await Promise.resolve(); + expect(registry.isPaused('item-a')).toBe(true); + + latePaused.resolve(); + await Promise.all([pausing, resuming]); + + expect(first.resume).toHaveBeenCalledOnce(); + expect(late.resume).toHaveBeenCalledOnce(); + expect(registry.isPaused('item-a')).toBe(false); + }); + it.each(['merge', 'split'] as const)('waits for %s termination before removing partial output', async (phase) => { const registry = new QueueProcessRegistry(); const closed = deferred(); diff --git a/src/main/queue/process-registry.ts b/src/main/queue/process-registry.ts index 1e54510..c4757fb 100644 --- a/src/main/queue/process-registry.ts +++ b/src/main/queue/process-registry.ts @@ -1,8 +1,25 @@ +import type { ChildProcess } from 'node:child_process'; + export type QueueProcessPhase = 'streamlink' | 'merge' | 'split' | 'post-processing'; +export function waitForChildProcessExit(process: ChildProcess | null, forceKillAfterMs = 5000): Promise { + if (!process || process.exitCode !== null || process.signalCode !== null) return Promise.resolve(); + return new Promise((resolve) => { + const finish = (): void => { + clearTimeout(timer); + resolve(); + }; + const timer = setTimeout(() => { + if (process.exitCode !== null || process.signalCode !== null) return; + try { process.kill('SIGKILL'); } catch { } + }, forceKillAfterMs); + process.once('close', finish); + }); +} + export interface QueueProcessResource { kill?: () => unknown; - wait?: Promise; + wait?: () => Promise; pause?: () => unknown | Promise; resume?: () => unknown | Promise; cancel?: () => unknown | Promise; @@ -27,6 +44,7 @@ export class QueueProcessRegistry { private readonly pausedItems = new Set(); private readonly cancellationWaiters = new Map void>>(); private readonly resumeWaiters = new Map void>>(); + private readonly pauseRuns = new Map>(); private readonly settling = new Set>(); private shuttingDown = false; @@ -43,6 +61,7 @@ export class QueueProcessRegistry { this.groups.set(itemId, group); } group.add(entry); + if (this.pausedItems.has(itemId)) this.enqueuePause(itemId, [entry]); return { accepted: true, @@ -52,13 +71,21 @@ export class QueueProcessRegistry { async pauseItem(itemId: string): Promise { this.pausedItems.add(itemId); - await this.invokeItem(itemId, 'pause'); + await this.enqueuePause(itemId, [...(this.groups.get(itemId) || [])]); } async resumeItem(itemId: string): Promise { - await this.invokeItem(itemId, 'resume'); - this.pausedItems.delete(itemId); - this.resolveWaiters(this.resumeWaiters, itemId); + while (this.pausedItems.has(itemId) && !this.cancelledItems.has(itemId)) { + const pauseRun = this.pauseRuns.get(itemId); + if (pauseRun) await pauseRun; + if (!this.pausedItems.has(itemId) || this.cancelledItems.has(itemId)) return; + if (pauseRun !== this.pauseRuns.get(itemId)) continue; + await this.invokeItem(itemId, 'resume'); + if (pauseRun !== this.pauseRuns.get(itemId)) continue; + this.pausedItems.delete(itemId); + this.pauseRuns.delete(itemId); + this.resolveWaiters(this.resumeWaiters, itemId); + } } async cancelItem(itemId: string): Promise { @@ -75,6 +102,7 @@ export class QueueProcessRegistry { this.groups.delete(itemId); this.cancellationWaiters.delete(itemId); this.resumeWaiters.delete(itemId); + this.pauseRuns.delete(itemId); this.pausedItems.delete(itemId); } @@ -86,6 +114,7 @@ export class QueueProcessRegistry { this.pausedItems.delete(itemId); this.cancellationWaiters.delete(itemId); this.resumeWaiters.delete(itemId); + this.pauseRuns.delete(itemId); } isCancelled(itemId: string): boolean { @@ -138,7 +167,7 @@ export class QueueProcessRegistry { entry.stopping = (async () => { try { entry.resource.kill?.(); } catch { } try { await entry.resource.cancel?.(); } catch { } - try { await entry.resource.wait; } catch { } + try { await entry.resource.wait?.(); } catch { } try { await entry.resource.cleanup?.(); } catch { } this.release(entry); })(); @@ -150,6 +179,18 @@ export class QueueProcessRegistry { void settlement.finally(() => this.settling.delete(settlement)); } + private enqueuePause(itemId: string, entries: RegisteredResource[]): Promise { + const previous = this.pauseRuns.get(itemId) || Promise.resolve(); + const pauseRun = Promise.allSettled([ + previous, + ...entries.map(async ({ resource }) => { + await resource.pause?.(); + }), + ]).then(() => undefined); + this.pauseRuns.set(itemId, pauseRun); + return pauseRun; + } + private release(entry: RegisteredResource): void { const group = this.groups.get(entry.itemId); if (!group) return;