fix(queue): await real process exit before cleanup

Start process-exit waits only when pause or cancellation begins, escalate stubborn children without resolving before close, latch pause across late registrations and fast resume, and retry concat only after a complete resume. Add real child-process and filesystem integration coverage for pause, shutdown cleanup, and persistence order.
2026-08-11 22:28:35 +02:00
parent 1140c8526e
commit de05539422
4 changed files with 235 additions and 67 deletions
@@ -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<typeof spawn>): Promise<void> {
+    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 });
+        }
+    });
+});
@@ -12,7 +12,7 @@ function deferred(): { promise: Promise<void>; resolve: () => void } {
 function createResource(wait: Promise<void> = 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();
@@ -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<void> {
+    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<unknown>;
+    wait?: () => Promise<unknown>;
    pause?: () => unknown | Promise<unknown>;
    resume?: () => unknown | Promise<unknown>;
    cancel?: () => unknown | Promise<unknown>;
@@ -27,6 +44,7 @@ export class QueueProcessRegistry {
    private readonly pausedItems = new Set<string>();
    private readonly cancellationWaiters = new Map<string, Set<() => void>>();
    private readonly resumeWaiters = new Map<string, Set<() => void>>();
+    private readonly pauseRuns = new Map<string, Promise<void>>();
    private readonly settling = new Set<Promise<void>>();
    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<void> {
        this.pausedItems.add(itemId);
-        await this.invokeItem(itemId, 'pause');
+        await this.enqueuePause(itemId, [...(this.groups.get(itemId) || [])]);
    }

    async resumeItem(itemId: string): Promise<void> {
-        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<void> {
@@ -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<void> {
+        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;