fix(cancel): wait for the worker to exit instead of forcing it after 500ms

app/lib/services/worker_manager.dart

@@ -156,31 +156,76 @@ class WorkerManager {
     }
   }
 
+  /// How long to let the worker shut itself down before forcing it.
+  ///
+  /// This MUST comfortably exceed the worker's cancellation poll interval
+  /// (`progress_interval`, 500ms in `pipeline_executor.rs`), because SIGTERM
+  /// only sets an atomic flag there — the actual teardown happens the next time
+  /// the progress loop comes round, and only then does it get to kill vspipe
+  /// and ffmpeg and reap them.
+  ///
+  /// It used to be exactly 500ms, i.e. precisely the poll interval, so the
+  /// worker essentially never won the race: it was SIGKILLed before reaching the
+  /// check. SIGKILL cannot be caught, so `PipelineExecutor::terminate()` (and its
+  /// `Drop`) never ran and vspipe/ffmpeg were reparented to init — left encoding
+  /// a cancelled job at full tilt, still writing to the output file, while the UI
+  /// reported "Job cancelled by user". Observed in the wild: three orphans at
+  /// ~670% CPU eleven minutes after a cancel, output past 320MB.
+  static const Duration _shutdownGrace = Duration(seconds: 5);
+
+  /// How long to wait for a SIGKILLed process to actually disappear.
+  static const Duration _forceKillGrace = Duration(seconds: 3);
+
   /// Cancels the current job.
+  ///
+  /// Waits for the worker to genuinely exit rather than assuming it has, so its
+  /// children are torn down by the worker itself. Only escalates to SIGKILL if
+  /// it is still alive after [_shutdownGrace].
   Future<void> cancel() async {
-    if (_process == null) return;
+    // Hold a local reference: `_cleanup()` nulls the field, and the old code
+    // tested `_process != null` *before* that ran, so its "force kill if still
+    // running" check was never actually false.
+    final process = _process;
+    if (process == null) return;
 
-    // Send SIGTERM on Unix, taskkill on Windows
     if (Platform.isWindows) {
-      // On Windows, we need to kill the process tree
-      await Process.run('taskkill', ['/PID', '${_process!.pid}', '/T', '/F']);
+      // No SIGTERM on Windows, and Process.kill maps to TerminateProcess, which
+      // does not touch children. taskkill /T walks the tree, so nothing is
+      // orphaned; /F is unavoidable there.
+      await Process.run('taskkill', ['/PID', '${process.pid}', '/T', '/F']);
     } else {
-      _process!.kill(ProcessSignal.sigterm);
+      process.kill(ProcessSignal.sigterm);
     }
 
-    // Give it a moment to clean up
-    await Future.delayed(const Duration(milliseconds: 500));
+    // Wait for the process to actually exit. `exitCode` completes once it has
+    // been reaped, so this is a real observation rather than a guess.
+    var exited = true;
+    try {
+      await process.exitCode.timeout(_shutdownGrace);
+    } on TimeoutException {
+      exited = false;
+    }
 
-    // Force kill if still running
-    if (_process != null) {
-      _process!.kill(ProcessSignal.sigkill);
+    if (!exited) {
+      // Genuinely wedged. Forcing it here orphans the children — the same
+      // failure described above — but by now the alternative is a job that
+      // never stops at all, so take the lesser problem and say so.
+      process.kill(ProcessSignal.sigkill);
+      try {
+        await process.exitCode.timeout(_forceKillGrace);
+      } on TimeoutException {
+        // Nothing further we can do from here.
+      }
     }
 
     _cleanup();
 
-    _emitCompletion(const CompletionResult(
+    _emitCompletion(CompletionResult(
       success: false,
-      errorMessage: 'Job cancelled by user',
+      errorMessage: exited
+          ? 'Job cancelled by user'
+          : 'Job cancelled by user (the worker had to be forced, so stray '
+              'ffmpeg/vspipe processes may still be running)',
       cancelled: true,
     ));
   }

app/test/cancel_shutdown_grace_test.dart

@@ -0,0 +1,103 @@
+// The cancel grace period and the worker's cancellation poll interval are a
+// cross-file coupling, and getting it wrong is silent.
+//
+// SIGTERM does not tear the pipeline down by itself: the worker's signal handler
+// only sets an atomic flag, and the actual teardown (killing vspipe and ffmpeg,
+// then reaping them) happens the next time the progress loop comes round —
+// `progress_interval` later, at most.
+//
+// The app used to wait exactly 500ms before SIGKILL, which is precisely that
+// poll interval, so the worker essentially never reached the check in time.
+// SIGKILL cannot be caught, so `PipelineExecutor::terminate()` and its `Drop`
+// never ran and vspipe/ffmpeg were reparented to init — left encoding a job the
+// user had cancelled, still writing to the output file, while the UI reported
+// "Job cancelled by user". Observed in the wild: three orphaned processes at
+// ~670% CPU eleven minutes after the cancel, output past 320MB.
+//
+// Nothing else catches this. The app reports a successful cancellation either
+// way, so the bug is invisible from inside the app — you have to look at the
+// process table. Hence a direct assertion that the two values stay in step.
+//
+// This is the same class of guard as `test_native_formats_match_pipe_source`
+// (Rust `NATIVE_FORMATS` vs Python `_FORMAT_MAP`).
+
+import 'dart:io';
+
+import 'package:path/path.dart' as p;
+import 'package:test/test.dart';
+
+String _repoRoot() {
+  var dir = Directory.current;
+  while (true) {
+    if (Directory(p.join(dir.path, 'worker')).existsSync() &&
+        Directory(p.join(dir.path, 'app')).existsSync()) {
+      return dir.path;
+    }
+    final parent = dir.parent;
+    if (parent.path == dir.path) {
+      throw StateError('could not locate the repo root from ${Directory.current}');
+    }
+    dir = parent;
+  }
+}
+
+void main() {
+  group('cancel shutdown grace', () {
+    late String root;
+
+    setUpAll(() => root = _repoRoot());
+
+    test('the app gives the worker longer than its cancellation poll interval',
+        () {
+      final rs = File(p.join(root, 'worker', 'src', 'pipeline_executor.rs'))
+          .readAsStringSync();
+      final dart = File(p.join(
+              root, 'app', 'lib', 'services', 'worker_manager.dart'))
+          .readAsStringSync();
+
+      final pollMatch = RegExp(
+        r'let\s+progress_interval\s*=\s*Duration::from_millis\((\d+)\)',
+      ).firstMatch(rs);
+      expect(pollMatch, isNotNull,
+          reason: 'could not find progress_interval in pipeline_executor.rs — '
+              'if it was renamed, update this test rather than deleting it');
+      final pollMs = int.parse(pollMatch!.group(1)!);
+
+      final graceMatch = RegExp(
+        r'_shutdownGrace\s*=\s*Duration\(seconds:\s*(\d+)\)',
+      ).firstMatch(dart);
+      expect(graceMatch, isNotNull,
+          reason: 'could not find _shutdownGrace in worker_manager.dart');
+      final graceMs = int.parse(graceMatch!.group(1)!) * 1000;
+
+      // The worker needs at least one full poll to notice the flag, then has to
+      // kill three children and reap them. A grace equal to (or barely above)
+      // the poll interval is the bug this test exists for, so require real
+      // headroom rather than a strict >.
+      expect(
+        graceMs,
+        greaterThanOrEqualTo(pollMs * 4),
+        reason: 'the cancel grace ($graceMs ms) must comfortably exceed the '
+            "worker's $pollMs ms cancellation poll, or SIGKILL wins the race "
+            'and vspipe/ffmpeg are orphaned mid-encode',
+      );
+    });
+
+    test('cancel waits for the process to exit instead of sleeping', () {
+      final dart = File(p.join(
+              root, 'app', 'lib', 'services', 'worker_manager.dart'))
+          .readAsStringSync();
+      final cancelStart = dart.indexOf('Future<void> cancel()');
+      expect(cancelStart, greaterThan(-1));
+      final body = dart.substring(cancelStart, dart.indexOf('_cleanup();', cancelStart));
+
+      // It must observe the exit, not assume it after a fixed delay.
+      expect(body, contains('exitCode'),
+          reason: 'cancel() must await the process exit; a fixed delay is what '
+              'orphaned the pipeline');
+      expect(body.contains('Future.delayed'), isFalse,
+          reason: 'cancel() must not gate the force-kill on a fixed delay — '
+              'wait on exitCode with a timeout instead');
+    });
+  });
+}

app/test/integration_cancel_test.dart

@@ -0,0 +1,191 @@
+// The worker must shut its pipeline down on SIGTERM, and do it quickly.
+//
+// `WorkerManager.cancel()` sends SIGTERM and then waits for the worker to exit
+// rather than forcing it after a fixed delay. That is only correct if the worker
+// really does tear down vspipe and ffmpeg and exit well inside the grace period,
+// so this test pins that contract: SIGTERM a running job, then assert it exits
+// in under `WorkerManager._shutdownGrace` with none of its children left behind.
+//
+// What this test does NOT do — and it is worth being exact, because the comment
+// it replaced claimed otherwise — is reproduce the orphaning bug itself. Killing
+// the worker with SIGKILL here still leaves no survivors: the children's stderr
+// pipes close with it and they die on EPIPE the next time they write. The
+// reported incident escaped that because the job was reading a large file off a
+// NAS, so the children sat blocked on I/O for minutes without writing anything,
+// long enough to be noticed at ~670% CPU.
+//
+// So the regression guard for the fix itself is `cancel_shutdown_grace_test`,
+// which fails against the old code. This test guards the assumption that fix
+// rests on. Do not weaken it into a "cancel returns without error" check.
+//
+// POSIX only — Windows has no SIGTERM, and the app uses `taskkill /T` there,
+// which kills the tree outright.
+@Tags(['heavy'])
+library;
+
+import 'dart:convert';
+import 'dart:io';
+
+import 'package:path/path.dart' as p;
+import 'package:test/test.dart';
+import 'package:uuid/uuid.dart';
+import 'package:vapourbox/models/encoding_settings.dart';
+import 'package:vapourbox/models/processing_pipeline.dart';
+import 'package:vapourbox/models/qtgmc_parameters.dart';
+import 'package:vapourbox/models/video_job.dart';
+
+import 'support/worker_harness.dart';
+
+/// PIDs of live processes whose executable path sits under [dir].
+Future<List<int>> _processesUnder(String dir) async {
+  final ps = await Process.run('ps', ['-Ao', 'pid=,command=']);
+  final out = <int>[];
+  for (final line in const LineSplitter().convert(ps.stdout.toString())) {
+    final trimmed = line.trimLeft();
+    final sp = trimmed.indexOf(' ');
+    if (sp <= 0) continue;
+    final pid = int.tryParse(trimmed.substring(0, sp));
+    if (pid == null) continue;
+    final cmd = trimmed.substring(sp + 1);
+    // Match only the executable path, not an argument that happens to name the
+    // deps dir (the job config and output paths can both mention it).
+    if (cmd.startsWith(dir)) out.add(pid);
+  }
+  return out;
+}
+
+void main() {
+  group('cancelling a job', () {
+    late String longInput;
+
+    setUpAll(() async {
+      await WorkerHarness.ensureReady();
+      await Directory(WorkerHarness.outputDir).create(recursive: true);
+
+      // The committed fixtures are only a few seconds long — short enough that
+      // QTGMC can finish before the cancel lands, which would satisfy every
+      // assertion below for the wrong reason. Build a source with minutes of
+      // work left in it, so the `exitCode != 0` guard genuinely means "still
+      // encoding when we cancelled it".
+      longInput = p.join(WorkerHarness.outputDir, 'cancel_long_source.avi');
+      if (!File(longInput).existsSync()) {
+        final gen = await Process.run(WorkerHarness.ffmpegPath, [
+          '-f', 'lavfi',
+          '-i', 'testsrc2=duration=60:size=720x576:rate=25',
+          '-vf', 'interlace',
+          '-c:v', 'ffv1', '-flags', '+ilme+ildct',
+          '-y', longInput,
+        ]);
+        expect(gen.exitCode, 0, reason: 'could not build the test source: ${gen.stderr}');
+      }
+    });
+
+    test('SIGTERM stops the worker and leaves no orphaned children', () async {
+      final depsDir = WorkerHarness.depsDir;
+      final outPath =
+          p.join(WorkerHarness.outputDir, 'test_cancel_orphans.mkv');
+
+      // A deliberately slow preset on a long source, so there is plenty of work
+      // outstanding at the moment we cancel.
+      final job = VideoJob(
+        id: const Uuid().v4(),
+        inputPath: longInput,
+        outputPath: outPath,
+        processingPipeline: const ProcessingPipeline(
+          deinterlace: QTGMCParameters(
+            enabled: true,
+            preset: QTGMCPreset.slower,
+            tff: true,
+          ),
+        ),
+        encodingSettings: const EncodingSettings(
+          codec: VideoCodec.h264,
+          container: ContainerFormat.mkv,
+          audioMode: AudioMode.passthrough,
+        ),
+      );
+
+      final configFile =
+          File(p.join(Directory.systemTemp.path, 'vb_cancel_orphans.json'));
+      await configFile.writeAsString(jsonEncode(job.toJson()));
+      addTearDown(() async {
+        if (await configFile.exists()) await configFile.delete();
+        final o = File(outPath);
+        if (await o.exists()) await o.delete();
+      });
+
+      final before = await _processesUnder(depsDir);
+
+      final proc = await Process.start(
+        WorkerHarness.workerPath,
+        ['--config', configFile.path],
+        environment: WorkerHarness.workerEnv,
+        workingDirectory: File(WorkerHarness.workerPath).parent.path,
+      );
+      proc.stdout.drain<void>();
+      proc.stderr.drain<void>();
+
+      // Wait until the pipeline is genuinely up, so we are not cancelling before
+      // any children exist — that would pass trivially.
+      var spawned = <int>[];
+      final deadline = DateTime.now().add(const Duration(seconds: 40));
+      while (DateTime.now().isBefore(deadline)) {
+        await Future<void>.delayed(const Duration(milliseconds: 500));
+        spawned = (await _processesUnder(depsDir))
+            .where((pid) => !before.contains(pid))
+            .toList();
+        if (spawned.length >= 2) break; // vspipe + at least one ffmpeg
+      }
+      expect(spawned.length, greaterThanOrEqualTo(2),
+          reason: 'the pipeline never started, so this would prove nothing');
+
+      // The app's cancel: SIGTERM, then wait for a real exit.
+      final sw = Stopwatch()..start();
+      proc.kill(ProcessSignal.sigterm);
+      final code = await proc.exitCode.timeout(
+        const Duration(seconds: 15),
+        onTimeout: () {
+          proc.kill(ProcessSignal.sigkill);
+          return -1;
+        },
+      );
+      sw.stop();
+
+      expect(code, isNot(-1),
+          reason: 'the worker ignored SIGTERM entirely');
+      // If the encode had simply finished, there would be nothing left to
+      // orphan and the check below would pass for the wrong reason. A cancelled
+      // worker exits 130 (or dies by signal); a completed one exits 0.
+      expect(code, isNot(0),
+          reason: 'the job completed before the cancel landed, so this run '
+              'proves nothing about orphaned children — lengthen the source');
+
+      // Children are reaped by the worker as it goes down; allow a moment for
+      // the process table to settle.
+      List<int> leftovers = [];
+      for (var i = 0; i < 20; i++) {
+        await Future<void>.delayed(const Duration(milliseconds: 500));
+        final now = await _processesUnder(depsDir);
+        leftovers = spawned.where(now.contains).toList();
+        if (leftovers.isEmpty) break;
+      }
+
+      // If this fails, kill them — a failing test must not leave the machine
+      // pinned at full CPU, which is the very problem under test.
+      if (leftovers.isNotEmpty) {
+        for (final pid in leftovers) {
+          Process.killPid(pid, ProcessSignal.sigkill);
+        }
+      }
+
+      expect(leftovers, isEmpty,
+          reason: 'vspipe/ffmpeg outlived a SIGTERMed worker; cancel() relies '
+              'on the worker reaping them, so this breaks the fix');
+
+      // The worker should go down well inside the app's 5s grace.
+      expect(sw.elapsed, lessThan(const Duration(seconds: 5)),
+          reason: 'worker took ${sw.elapsed.inMilliseconds}ms to shut down, '
+              'which does not fit inside WorkerManager._shutdownGrace');
+    }, timeout: const Timeout(Duration(minutes: 3)));
+  });
+}