From f2539bed66d57828bfdfaaae64cefc49383d456b Mon Sep 17 00:00:00 2001 From: Stuart Cameron Date: Fri, 7 Aug 2026 22:38:42 +1000 Subject: [PATCH] fix(cancel): wait for the worker to exit instead of forcing it after 500ms Cancelling a job left vspipe and ffmpeg running. Reported after a real cancel: three orphaned processes at ~670% CPU eleven minutes later, still writing to the output file of a job the user had cancelled, while the UI showed "Job cancelled by user". The worker side was already correct -- on cancel it calls PipelineExecutor::terminate() and has a Drop impl doing the same. The bug was that it never got the chance. WorkerManager.cancel() sent SIGTERM, slept a flat 500ms, then SIGKILLed. The worker's signal handler only sets an atomic flag; the teardown happens the next time the progress loop comes round, and that loop sleeps on progress_interval -- which is also 500ms. So the grace period was exactly the poll interval and the worker essentially never reached the check in time. SIGKILL cannot be caught, so terminate() and Drop never ran and the children were reparented to init. The "force kill if still running" guard did not help either: it tested `_process != null`, but `_process` is only nulled by _cleanup(), which runs afterwards -- so the SIGKILL was unconditional. Now cancel() awaits the process's actual exitCode with a 5s timeout and only escalates to SIGKILL if it is genuinely still alive, reporting that in the completion message so a forced kill is not silently indistinguishable from a clean one. Windows keeps taskkill /T, which walks the tree and so cannot orphan. Tests: - cancel_shutdown_grace_test (per-push) pins the coupling that caused this: the grace must be several times the worker's poll interval, and cancel() must wait on exitCode rather than a fixed delay. Both assertions fail against the old code, verified by reverting to it. - integration_cancel_test (heavy) pins the contract the fix depends on: SIGTERM a running job and assert the worker exits inside the grace with no children left. It builds a 60s source, because the committed fixtures are short enough that QTGMC can finish before the cancel lands and satisfy everything for the wrong reason. Worth recording what that second test does NOT do: it cannot reproduce the orphaning. SIGKILLing the worker in this harness still leaves no survivors, because the children's stderr pipes close with it and they die on EPIPE at the next write. The reported incident escaped that only because the job was reading a large file off a NAS and the children sat blocked on I/O for minutes without writing. The per-push guards are therefore the real regression protection. --- app/lib/services/worker_manager.dart | 69 ++++++-- app/test/cancel_shutdown_grace_test.dart | 103 ++++++++++++ app/test/integration_cancel_test.dart | 191 +++++++++++++++++++++++ 3 files changed, 351 insertions(+), 12 deletions(-) create mode 100644 app/test/cancel_shutdown_grace_test.dart create mode 100644 app/test/integration_cancel_test.dart diff --git a/app/lib/services/worker_manager.dart b/app/lib/services/worker_manager.dart index 7561c33..37184ee 100644 --- a/app/lib/services/worker_manager.dart +++ b/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 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, )); } diff --git a/app/test/cancel_shutdown_grace_test.dart b/app/test/cancel_shutdown_grace_test.dart new file mode 100644 index 0000000..6163549 --- /dev/null +++ b/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 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'); + }); + }); +} diff --git a/app/test/integration_cancel_test.dart b/app/test/integration_cancel_test.dart new file mode 100644 index 0000000..47ef8e7 --- /dev/null +++ b/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> _processesUnder(String dir) async { + final ps = await Process.run('ps', ['-Ao', 'pid=,command=']); + final out = []; + 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(); + proc.stderr.drain(); + + // Wait until the pipeline is genuinely up, so we are not cancelling before + // any children exist — that would pass trivially. + var spawned = []; + final deadline = DateTime.now().add(const Duration(seconds: 40)); + while (DateTime.now().isBefore(deadline)) { + await Future.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 leftovers = []; + for (var i = 0; i < 20; i++) { + await Future.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))); + }); +}