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))); + }); +}