diff --git a/app/lib/services/job_runner.dart b/app/lib/services/job_runner.dart new file mode 100644 index 0000000..a87455a --- /dev/null +++ b/app/lib/services/job_runner.dart @@ -0,0 +1,48 @@ +import '../models/progress_info.dart'; +import '../models/video_job.dart'; +import 'worker_manager.dart' show CompletionResult, LogMessage; + +/// What [MainViewModel] needs from the thing that runs jobs. +/// +/// This exists as a seam for testing, and the reason is worth stating: three +/// separate fixes for the same cancellation hang shipped without ever being +/// exercised, because `MainViewModel` constructed its `WorkerManager` directly +/// and so the queue state machine could not be driven from a test at all. Every +/// fix rested on reading the code, and two of them were "verified" by tests that +/// only read the source text. +/// +/// The interface is deliberately the *observed* surface — the streams the +/// viewmodel listens to and the three calls it makes — not everything +/// WorkerManager can do. Anything outside this (GPU probing, process groups) +/// stays on the concrete class. +abstract interface class JobRunner { + /// Progress updates for the running job. + Stream get progressStream; + + /// Log lines from the worker. + Stream get logStream; + + /// Exactly one event per started job — see [startJob]. + Stream get completionStream; + + /// Whether a job is currently running. + bool get isRunning; + + /// Starts [job]. + /// + /// Implementations must emit exactly one [CompletionResult] for every call + /// that returns normally: no more (the UI would act twice) and no fewer (the + /// UI latches on `cancelling`/`processing` and never returns to idle, which is + /// the hang this interface was introduced to make testable). + Future startJob(VideoJob job); + + /// Cancels the running job. + /// + /// Must still emit a completion when there is nothing to kill — the caller has + /// already moved the UI into `cancelling` by the time this is called, so + /// emitting nothing strands it there with Start and Cancel both disabled. + Future cancel(); + + /// Releases resources. + void dispose(); +} diff --git a/app/lib/services/tool_locator.dart b/app/lib/services/tool_locator.dart index 6e434c4..3e0a4c2 100644 --- a/app/lib/services/tool_locator.dart +++ b/app/lib/services/tool_locator.dart @@ -111,6 +111,16 @@ class ToolLocator { /// Resolve the vapourbox-worker executable path. String? _resolveWorker() { + // Explicit override, mirroring VAPOURBOX_DEPS_DIR above. Under `flutter + // test` the resolved executable is the test runner, not the app bundle, so + // neither the production nor the dev path below can find the worker — which + // left WorkerManager, and therefore the whole cancel/restart flow, with no + // way to be tested against a real worker at all. + final override = Platform.environment['VAPOURBOX_WORKER']; + if (override != null && override.isNotEmpty && File(override).existsSync()) { + return override; + } + final exeDir = path.dirname(Platform.resolvedExecutable); final ext = Platform.isWindows ? '.exe' : ''; final workerExe = 'vapourbox-worker$ext'; diff --git a/app/lib/services/worker_manager.dart b/app/lib/services/worker_manager.dart index 5508648..cf8d889 100644 --- a/app/lib/services/worker_manager.dart +++ b/app/lib/services/worker_manager.dart @@ -2,14 +2,17 @@ import 'dart:async'; import 'dart:convert'; import 'dart:io'; +import 'package:flutter/foundation.dart'; + import '../models/progress_info.dart'; import '../models/video_job.dart'; +import 'job_runner.dart'; import 'process_tree.dart'; import 'temp_directory_service.dart'; import 'tool_locator.dart'; /// Manages the worker process lifecycle and IPC. -class WorkerManager { +class WorkerManager implements JobRunner { Process? _process; StreamSubscription? _stdoutSubscription; StreamSubscription? _stderrSubscription; @@ -51,24 +54,38 @@ class WorkerManager { /// "Worker exited with code 143". bool _cancelRequested = false; + /// Whether a started job still owes a [CompletionResult]. + /// + /// `_completionEmitted` enforces "at most one" completion per job. This + /// enforces the other half — "at least one" — which nothing did before, and + /// whose absence is the hang: `cancel()` returned silently when there was no + /// process, the viewmodel had already latched `_state = cancelling`, and with + /// no event to retire that latch the UI sat with Start *and* Cancel disabled. + bool _jobInFlight = false; + /// Stream of progress updates from the worker. final _progressController = StreamController.broadcast(); + @override Stream get progressStream => _progressController.stream; /// Stream of log messages from the worker. final _logController = StreamController.broadcast(); + @override Stream get logStream => _logController.stream; /// Stream of completion events. final _completionController = StreamController.broadcast(); + @override Stream get completionStream => _completionController.stream; /// Whether the worker is currently running. + @override bool get isRunning => _process != null; /// Starts a deinterlacing job. /// /// Creates a temporary JSON config file and spawns the worker process. + @override Future startJob(VideoJob job) async { if (_process != null) { throw StateError('Worker is already running'); @@ -76,6 +93,7 @@ class WorkerManager { _completionEmitted = false; _cancelRequested = false; + _jobInFlight = true; final generation = ++_generation; final toolLocator = ToolLocator.instance; @@ -99,6 +117,7 @@ class WorkerManager { workingDirectory: File(workerPath).parent.path, ); + // Listen to stdout for JSON messages _stdoutSubscription = _process!.stdout .transform(utf8.decoder) @@ -220,17 +239,40 @@ class WorkerManager { /// 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]. + @override Future cancel() async { // 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; // Which job this call is cancelling. Everything after the await below is // conditional on it still being the current one. final generation = _generation; _cancelRequested = true; + + if (process == null) { + // Nothing to signal — the job finished microseconds ago, `startJob` threw, + // or we are still inside `preparingJob` and the process does not exist + // yet. That last case is reachable from the UI: `canCancel` includes + // `preparingJob`, so the Cancel button is live before there is anything to + // cancel. + // + // The caller has already moved the UI into `cancelling`, so returning + // silently here is what stranded it: `cancelling` is neither `idle` nor in + // `canCancel`, leaving Start and Cancel both disabled with no way back. + // Report the cancellation so the latch can be retired. + if (_jobInFlight) { + _emitCompletion(const CompletionResult( + success: false, + cancelled: true, + errorMessage: 'Job cancelled by user', + )); + _cleanup(); + } + return; + } + if (Platform.isWindows) { // No SIGTERM on Windows, and Process.kill maps to TerminateProcess, which // does not touch children. taskkill /T walks the tree, so nothing is @@ -310,7 +352,14 @@ class WorkerManager { required String? lastError, required int exitCode, }) { - if (cancelRequested) { + // Cancellation is observed two ways, and either is sufficient. + // + // `cancelRequested` is what the app asked for. Exit 130 is what the worker + // actually did (`worker/src/main.rs` returns it once the cancel flag is + // seen). Taking both means a cancellation initiated anywhere — including a + // SIGINT the app never issued — is still reported as one, rather than being + // inferred solely from a flag the app sets for itself. + if (cancelRequested || exitCode == 130) { return const CompletionResult( success: false, cancelled: true, @@ -331,8 +380,11 @@ class WorkerManager { /// Emit [result] unless this job has already reported one. void _emitCompletion(CompletionResult result) { - if (_completionEmitted || _completionController.isClosed) return; + if (_completionEmitted || _completionController.isClosed) { + return; + } _completionEmitted = true; + _jobInFlight = false; _completionController.add(result); } @@ -396,6 +448,7 @@ class WorkerManager { } /// Disposes of resources. + @override void dispose() { cancel(); _progressController.close(); diff --git a/app/lib/viewmodels/main_viewmodel.dart b/app/lib/viewmodels/main_viewmodel.dart index 5778f23..ba906a7 100644 --- a/app/lib/viewmodels/main_viewmodel.dart +++ b/app/lib/viewmodels/main_viewmodel.dart @@ -21,11 +21,14 @@ import '../services/frame_math.dart'; import '../services/preset_service.dart'; import '../services/preview_generator.dart'; import '../services/temp_directory_service.dart'; +import '../services/job_runner.dart'; import '../services/worker_manager.dart'; /// Main view model managing application state. class MainViewModel extends ChangeNotifier { - final WorkerManager _workerManager = WorkerManager(); + /// Runs jobs. Injectable so the queue state machine can be driven from a + /// test — see JobRunner for why that matters. + final JobRunner _workerManager; final FieldOrderDetector _fieldOrderDetector = FieldOrderDetector(); final PreviewGenerator _previewGenerator = PreviewGenerator(); final DvdService _dvdService = DvdService(); @@ -337,7 +340,8 @@ class MainViewModel extends ChangeNotifier { return _manualFieldOrder; } - MainViewModel() { + MainViewModel({JobRunner? runner}) + : _workerManager = runner ?? WorkerManager() { _setupSubscriptions(); _initializePreviewGenerator(); _probeGpuCapabilities(); @@ -1088,11 +1092,10 @@ class MainViewModel extends ChangeNotifier { if (nextIndex == -1) { // No more items to process - _isQueueProcessing = false; - _currentProcessingIndex = -1; - // Show failed state if any items failed, otherwise completed final hasFailed = _queue.any((q) => q.status == QueueItemStatus.failed); - _state = hasFailed ? ProcessingState.failed : ProcessingState.completed; + _stopProcessing( + state: hasFailed ? ProcessingState.failed : ProcessingState.completed, + ); notifyListeners(); return; } @@ -1193,6 +1196,15 @@ class MainViewModel extends ChangeNotifier { /// Handles completion of a queue item. Future _handleQueueItemCompletion(CompletionResult result) async { if (_currentProcessingIndex < 0 || _currentProcessingIndex >= _queue.length) { + // A completion with no item to attribute it to. Dropping it silently is + // what poisons the app: both latches stay set, `canProcess` stays false + // and `startQueueProcessing` returns at its guard, so the UI spins with + // Start disabled and no way back short of a restart. + // + // The event is unattributable, not meaningless — the worker has stopped. + // Stand down rather than latch. + _stopProcessing(); + notifyListeners(); return; } @@ -1200,9 +1212,10 @@ class MainViewModel extends ChangeNotifier { if (result.cancelled) { item.status = QueueItemStatus.cancelled; - _isQueueProcessing = false; - _currentProcessingIndex = -1; - _state = ProcessingState.idle; + // Cancel stops the whole queue; remaining ready items stay ready for a + // later Start. _stopProcessing runs after the status is set, so this item + // is not caught by its rescue of stranded `processing` items. + _stopProcessing(); } else if (result.success) { item.status = QueueItemStatus.completed; // Process next item @@ -1223,6 +1236,24 @@ class MainViewModel extends ChangeNotifier { notifyListeners(); } + /// Return the queue to a state the user can act on. + /// + /// Every path that stops processing goes through here, so recovery cannot be + /// forgotten on one of them. In particular it rescues items stranded in + /// `processing`: that status is neither `canProcess` nor `canReprocess`, so + /// `startQueueProcessing`'s reset loop skips it and the item can never be run + /// again — the queue stays permanently unusable for that file. + void _stopProcessing({ProcessingState state = ProcessingState.idle}) { + for (final item in _queue) { + if (item.status == QueueItemStatus.processing) { + item.status = QueueItemStatus.ready; + } + } + _isQueueProcessing = false; + _currentProcessingIndex = -1; + _state = state; + } + /// Gets the effective field order for a queue item. FieldOrder _getEffectiveFieldOrder(QueueItem item) { if (_autoFieldOrder && item.videoInfo?.fieldOrder != null) { @@ -1255,6 +1286,29 @@ class MainViewModel extends ChangeNotifier { } else { await _workerManager.cancel(); } + + // A completion emitted during the cancel is delivered asynchronously + // (broadcast stream), so checking immediately races it — and winning that + // race is worse than losing it: the queue's own handler never runs, the item + // is never marked `cancelled`, and recovery happens by fallback instead of + // by design. Give delivery a turn of the event loop first. + await Future.delayed(Duration.zero); + await Future.delayed(Duration.zero); + + // By now a completion should have retired the `cancelling` latch. If one + // never arrived, stand down anyway: `cancelling` is neither `idle` nor in + // `canCancel`, so staying there disables Start *and* Cancel with no way back + // short of restarting the app. A missed event should cost the user a + // mislabelled item, not a dead window. + if (_state == ProcessingState.cancelling) { + _logMessages.add(LogMessage( + level: LogLevel.warning, + message: 'Cancellation completed without a result from the worker; ' + 'returning to idle.', + )); + _stopProcessing(); + notifyListeners(); + } } /// Retries all failed items in the queue. diff --git a/app/test/cancel_shutdown_grace_test.dart b/app/test/cancel_shutdown_grace_test.dart index 6163549..e1a2e10 100644 --- a/app/test/cancel_shutdown_grace_test.dart +++ b/app/test/cancel_shutdown_grace_test.dart @@ -21,6 +21,12 @@ // This is the same class of guard as `test_native_formats_match_pipe_source` // (Rust `NATIVE_FORMATS` vs Python `_FORMAT_MAP`). +// NOTE: always normalise line endings when reading source here. git checks these +// files out CRLF on Windows, and every scan below is written against "\n". This +// has broken CI three separate times — twice with a confusing "expected a value +// less than N" and once with a RangeError from indexOf returning -1. If a scan +// can be replaced by a behavioural assertion, prefer that; these read source +// only because the thing they guard is a code shape, not an observable. import 'dart:io'; import 'package:path/path.dart' as p; @@ -50,10 +56,12 @@ void main() { test('the app gives the worker longer than its cancellation poll interval', () { final rs = File(p.join(root, 'worker', 'src', 'pipeline_executor.rs')) - .readAsStringSync(); + .readAsStringSync() + .replaceAll('\r\n', '\n'); final dart = File(p.join( root, 'app', 'lib', 'services', 'worker_manager.dart')) - .readAsStringSync(); + .readAsStringSync() + .replaceAll('\r\n', '\n'); final pollMatch = RegExp( r'let\s+progress_interval\s*=\s*Duration::from_millis\((\d+)\)', @@ -86,10 +94,16 @@ void main() { test('cancel waits for the process to exit instead of sleeping', () { final dart = File(p.join( root, 'app', 'lib', 'services', 'worker_manager.dart')) - .readAsStringSync(); + .readAsStringSync() + .replaceAll('\r\n', '\n'); final cancelStart = dart.indexOf('Future cancel()'); expect(cancelStart, greaterThan(-1)); - final body = dart.substring(cancelStart, dart.indexOf('_cleanup();', cancelStart)); + // Read the whole method, not up to the first `_cleanup()`. cancel() now + // has an early branch for "nothing to kill" which cleans up before the + // await, so anchoring on the first occurrence pointed at that instead — + // a false failure with the behaviour perfectly correct. + final body = + dart.substring(cancelStart, dart.indexOf('\n }\n', cancelStart)); // It must observe the exit, not assume it after a fixed delay. expect(body, contains('exitCode'), diff --git a/app/test/fakes/fake_job_runner.dart b/app/test/fakes/fake_job_runner.dart new file mode 100644 index 0000000..0c68c97 --- /dev/null +++ b/app/test/fakes/fake_job_runner.dart @@ -0,0 +1,95 @@ +import 'dart:async'; + +import 'package:vapourbox/models/progress_info.dart'; +import 'package:vapourbox/models/video_job.dart'; +import 'package:vapourbox/services/job_runner.dart'; +import 'package:vapourbox/services/worker_manager.dart'; + +/// A [JobRunner] that runs nothing, so the queue state machine can be driven +/// deterministically. +/// +/// The point is to make the *absence* of events testable. Three fixes for the +/// same hang shipped unverified because nothing could ask "what happens when the +/// completion never arrives?" — see [emitNothingOnCancel]. +class FakeJobRunner implements JobRunner { + final _progress = StreamController.broadcast(); + final _log = StreamController.broadcast(); + final _completion = StreamController.broadcast(); + + /// Every job handed to [startJob], in order. + final List started = []; + + /// How many times [cancel] was called. + int cancelCount = 0; + + /// Whether a job is notionally running. + bool _running = false; + + /// Make [cancel] emit no completion at all. + /// + /// This is the shipped bug: `WorkerManager.cancel()` returned early when + /// `_process` was null and reported nothing, while the viewmodel had already + /// latched `_state = cancelling`. The UI then had Start and Cancel both + /// disabled with no way back. + bool emitNothingOnCancel = false; + + /// Make [startJob] throw, as it does when the worker binary is missing. + bool failToStart = false; + + @override + Stream get progressStream => _progress.stream; + + @override + Stream get logStream => _log.stream; + + @override + Stream get completionStream => _completion.stream; + + @override + bool get isRunning => _running; + + @override + Future startJob(VideoJob job) async { + if (failToStart) throw Exception('worker not found'); + started.add(job); + _running = true; + } + + @override + Future cancel() async { + cancelCount++; + _running = false; + if (emitNothingOnCancel) return; + complete(const CompletionResult( + success: false, + cancelled: true, + errorMessage: 'Job cancelled by user', + )); + } + + @override + void dispose() { + _progress.close(); + _log.close(); + _completion.close(); + } + + /// Emit a completion as the real runner would, and let the viewmodel's + /// listener run before returning. + Future complete(CompletionResult result) async { + _running = false; + _completion.add(result); + // Broadcast delivery is asynchronous; give the listener (and any + // _processNextItem it triggers) a chance to run. + await Future.delayed(Duration.zero); + await Future.delayed(Duration.zero); + } + + void emitProgress(ProgressInfo p) => _progress.add(p); + + Future succeed({String output = '/tmp/out.mkv'}) => + complete(CompletionResult(success: true, outputPath: output)); + + Future fail({String error = 'ffmpeg died'}) => + complete(CompletionResult(success: false, errorMessage: error)); +} diff --git a/app/test/integration_stale_progress_file_test.dart b/app/test/integration_stale_progress_file_test.dart new file mode 100644 index 0000000..40e4af5 --- /dev/null +++ b/app/test/integration_stale_progress_file_test.dart @@ -0,0 +1,137 @@ +// A leftover progress file must not convince the worker the encode is done. +// +// The worker polls `${TMPDIR}/vb_progress_${job.id}` for ffmpeg's progress, and +// `job.id` is the queue item's id — identical every time that item is re-run. On +// the way out, ffmpeg writes `progress=end`, so a cancelled run leaves one +// behind. The next run's loop polls immediately, before its own ffmpeg has +// opened and truncated the file, reads the previous run's tail, concludes the +// encode has already finished, breaks out on the first iteration, and then +// blocks forever in `decoder.wait()` while the pipeline encodes at full speed +// behind it. +// +// The user-visible result is a job that never reports progress and never +// completes: the app sits on "processing" with a spinner. Four fixes were made +// in the app for that symptom before the cause was found here, because the app +// was behaving correctly throughout — the worker genuinely never reported +// anything. +// +// It never reproduced under test because every test generated a fresh job id, so +// the file never pre-existed. This test seeds it deliberately, which is the only +// way to make the failure deterministic — in the wild it is a race between the +// first poll and ffmpeg's truncate, which is why it came and went. +@Tags(['heavy']) +library; + +import 'dart:convert'; +import 'dart:io'; + +import 'package:flutter_test/flutter_test.dart'; +import 'package:path/path.dart' as p; +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'; + +void main() { + setUpAll(() async => WorkerHarness.ensureReady()); + + test('a stale progress=end does not abort the next run', () async { + final jobId = const Uuid().v4(); + + // Exactly what a cancelled run leaves behind: real progress lines followed + // by the terminating `progress=end`. + final stale = File(p.join(Directory.systemTemp.path, 'vb_progress_$jobId')); + await stale.writeAsString([ + 'frame=1200', + 'fps=48.0', + 'out_time_ms=48000000', + 'dup_frames=0', + 'drop_frames=0', + 'speed=1.9x', + 'progress=end', + '', + ].join('\n')); + expect(await stale.exists(), isTrue); + addTearDown(() async { + if (await stale.exists()) await stale.delete(); + }); + + final job = VideoJob( + id: jobId, // the same id as the "previous run", as a re-run would be + inputPath: WorkerHarness.inputFile, + outputPath: p.join(WorkerHarness.outputDir, 'stale_progress.mkv'), + processingPipeline: const ProcessingPipeline( + deinterlace: + QTGMCParameters(enabled: true, preset: QTGMCPreset.fast, tff: true), + ), + encodingSettings: const EncodingSettings( + codec: VideoCodec.h264, + container: ContainerFormat.mkv, + audioMode: AudioMode.passthrough, + ), + ); + + final result = await WorkerHarness.runJob( + job.toJson(), + label: 'stale progress file', + timeout: const Duration(minutes: 3), + ); + + // Before the fix this timed out: the worker broke out of its progress loop + // on the first iteration and sat in decoder.wait() forever. + expect(result.success, isTrue, + reason: 'the worker did not complete with a stale progress file ' + 'present.\n${result.error}\n' + '${result.logs.length > 20 ? result.logs.sublist(result.logs.length - 20).join('\n') : result.logs.join('\n')}'); + + final out = File(result.outputPath!); + expect(await out.exists(), isTrue, reason: 'no output produced'); + expect(await out.length(), greaterThan(0)); + + // And it must have actually encoded, not just exited cleanly. + final v = await WorkerHarness.firstStream(result.outputPath!, + selector: 'v:0', entries: ['codec_name', 'width', 'height']); + expect(v, isNotNull, reason: 'output has no video stream'); + }, timeout: const Timeout(Duration(minutes: 5))); + + test('progress is still reported when a stale file is present', () async { + // The completion check above can pass on a short clip even if progress was + // never reported. Progress reaching the app is the actual symptom, so assert + // it directly. + final jobId = const Uuid().v4(); + final stale = File(p.join(Directory.systemTemp.path, 'vb_progress_$jobId')); + await stale.writeAsString('frame=999\nprogress=end\n'); + addTearDown(() async { + if (await stale.exists()) await stale.delete(); + }); + + final job = VideoJob( + id: jobId, + inputPath: WorkerHarness.inputFile, + outputPath: p.join(WorkerHarness.outputDir, 'stale_progress_events.mkv'), + processingPipeline: const ProcessingPipeline( + deinterlace: QTGMCParameters( + enabled: true, preset: QTGMCPreset.slow, tff: true), + ), + encodingSettings: const EncodingSettings( + codec: VideoCodec.h264, + container: ContainerFormat.mkv, + audioMode: AudioMode.passthrough, + ), + ); + + final result = await WorkerHarness.runJob(job.toJson(), + label: 'stale progress, events', timeout: const Duration(minutes: 3)); + + expect(result.success, isTrue, reason: result.error ?? 'job failed'); + final progressLines = result.logs + .where((l) => l.contains('"type":"progress"') || l.contains('Progress:')) + .length; + expect(progressLines, greaterThan(0), + reason: 'the job completed but never reported progress — the loop still ' + 'exited early and the UI would show nothing'); + }, timeout: const Timeout(Duration(minutes: 5))); +} diff --git a/app/test/integration_worker_manager_restart_test.dart b/app/test/integration_worker_manager_restart_test.dart new file mode 100644 index 0000000..4173126 --- /dev/null +++ b/app/test/integration_worker_manager_restart_test.dart @@ -0,0 +1,162 @@ +// WorkerManager must keep delivering progress after a cancel and restart. +// +// The viewmodel is exonerated for this: driven against a fake runner it relays +// progress perfectly across a cancel and a restart +// (`main_viewmodel_lifecycle_test`). The reported symptom — the second job runs +// but the UI never updates — therefore lives in WorkerManager's real process and +// stdout handling, which until now had no test at all, for the same structural +// reason the viewmodel had none: it could not be constructed with a reachable +// worker binary. +// +// Needs the real worker and deps, so it is heavy. +@Tags(['heavy']) +library; + +import 'dart:async'; +import 'dart:convert'; +import 'dart:io'; + +import 'package:flutter_test/flutter_test.dart'; +import 'package:path/path.dart' as p; +import 'package:uuid/uuid.dart'; +import 'package:vapourbox/models/encoding_settings.dart'; +import 'package:vapourbox/models/processing_pipeline.dart'; +import 'package:vapourbox/models/progress_info.dart'; +import 'package:vapourbox/models/qtgmc_parameters.dart'; +import 'package:vapourbox/models/video_job.dart'; +import 'package:vapourbox/services/tool_locator.dart'; +import 'package:vapourbox/services/worker_manager.dart'; + +import 'support/worker_harness.dart'; + +void main() { + late String longInput; + + setUpAll(() async { + await WorkerHarness.ensureReady(); + await Directory(WorkerHarness.outputDir).create(recursive: true); + + // ToolLocator resolves relative to the executable, which under `flutter + // test` is the test runner. Point it at the real deps and worker. + // (Set before initialize(); the values are cached.) + 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 source: ${gen.stderr}'); + } + await ToolLocator.instance.initialize(); + }); + + VideoJob jobFor(String name) => VideoJob( + id: const Uuid().v4(), + inputPath: longInput, + outputPath: p.join(WorkerHarness.outputDir, '$name.mkv'), + processingPipeline: const ProcessingPipeline( + deinterlace: + QTGMCParameters(enabled: true, preset: QTGMCPreset.fast, tff: true), + ), + encodingSettings: const EncodingSettings( + codec: VideoCodec.h264, + container: ContainerFormat.mkv, + audioMode: AudioMode.passthrough, + ), + ); + + /// Wait for [n] progress events, or give up. + Future> collectProgress( + Stream stream, { + int n = 1, + Duration timeout = const Duration(seconds: 60), + }) async { + final seen = []; + final done = Completer(); + final sub = stream.listen((e) { + seen.add(e); + if (seen.length >= n && !done.isCompleted) done.complete(); + }); + await done.future.timeout(timeout, onTimeout: () {}); + await sub.cancel(); + return seen; + } + + test('a restart that lands while cancel is still in flight still reports', + () async { + // The UI does not await cancelProcessing() (progress_panel.dart:440), so a + // user who cancels and immediately starts again races the cancel. That path + // is not covered by the awaited version below, and it is the one where + // startJob can find _process still set and throw "Worker is already + // running" — which surfaces as a job that never starts and a UI that never + // updates. + final wm = WorkerManager(); + addTearDown(wm.dispose); + + await wm.startJob(jobFor('inflight_first')); + await collectProgress(wm.progressStream, n: 1); + + // Fire and forget, exactly as the button does. + final cancelFuture = wm.cancel(); + + // Restart as soon as the manager says it is free, without awaiting cancel. + final deadline = DateTime.now().add(const Duration(seconds: 20)); + while (wm.isRunning && DateTime.now().isBefore(deadline)) { + await Future.delayed(const Duration(milliseconds: 50)); + } + + Object? startError; + try { + await wm.startJob(jobFor('inflight_second')); + } catch (e) { + startError = e; + } + expect(startError, isNull, + reason: 'restarting while the cancel was still settling threw: ' + '$startError — the job never starts and the UI never updates'); + + final progress = await collectProgress(wm.progressStream, n: 2); + expect(progress, isNotEmpty, + reason: 'the restarted job produced no progress'); + + await cancelFuture; + await wm.cancel(); + }, timeout: const Timeout(Duration(minutes: 5))); + + test('progress still flows after a cancel and a restart', () async { + expect(ToolLocator.instance.workerPath, isNotNull, + reason: 'set VAPOURBOX_WORKER (and VAPOURBOX_DEPS_DIR) for this test'); + + final wm = WorkerManager(); + addTearDown(wm.dispose); + + // --- first job: prove progress flows at all --- + final first = collectProgress(wm.progressStream, n: 2); + await wm.startJob(jobFor('restart_first')); + final firstProgress = await first; + expect(firstProgress, isNotEmpty, + reason: 'no progress from the first job — the harness itself is wrong'); + + // --- cancel --- + final cancelled = wm.completionStream.first; + await wm.cancel(); + final result = await cancelled.timeout(const Duration(seconds: 20)); + expect(result.cancelled, isTrue, reason: 'a cancel must report as cancelled'); + expect(wm.isRunning, isFalse, reason: 'cancel must leave nothing running'); + + // --- restart: this is the reported bug --- + final second = collectProgress(wm.progressStream, n: 2); + await wm.startJob(jobFor('restart_second')); + final secondProgress = await second; + + expect(secondProgress, isNotEmpty, + reason: 'the second job produced no progress events. The job runs, but ' + 'nothing reaches the UI — the stdout subscription is not feeding ' + 'the progress stream after a cancel.'); + + await wm.cancel(); + }, timeout: const Timeout(Duration(minutes: 5))); +} diff --git a/app/test/main_viewmodel_lifecycle_test.dart b/app/test/main_viewmodel_lifecycle_test.dart new file mode 100644 index 0000000..123863d --- /dev/null +++ b/app/test/main_viewmodel_lifecycle_test.dart @@ -0,0 +1,171 @@ +// The queue state machine, driven end to end against a fake runner. +// +// This file exists because three fixes for the same hang shipped without ever +// being exercised. `MainViewModel` constructed its own `WorkerManager`, so the +// state machine could not be driven from a test at all; every fix rested on +// reading the code, and two were "verified" by tests that only read source text. +// +// What makes the hang possible is that two independent latches gate everything — +// `_state` and `_isQueueProcessing` — and both are retired only by a +// `CompletionResult`. If that event does not arrive, `canProcess` stays false, +// `startQueueProcessing` returns at its guard, and the UI spins with Start and +// Cancel both disabled. So the important cases below are the ones where an event +// is missing or unattributable, not the happy path. + +import 'package:flutter_test/flutter_test.dart'; +import 'package:vapourbox/models/progress_info.dart'; +import 'package:vapourbox/models/queue_item.dart'; +import 'package:vapourbox/services/worker_manager.dart'; +import 'package:vapourbox/viewmodels/main_viewmodel.dart'; + +import 'fakes/fake_job_runner.dart'; + +void main() { + TestWidgetsFlutterBinding.ensureInitialized(); + + late FakeJobRunner runner; + late MainViewModel vm; + + /// A queue with [n] items sitting at `ready`. + /// + /// Analysis fails without real files, which the viewmodel tolerates — items + /// still end up `ready`, which is all these tests need. + Future seedQueue(int n) async { + await vm.addMultipleToQueue([ + for (var i = 0; i < n; i++) '/tmp/vapourbox_test_$i.mkv', + ]); + for (final item in vm.queue) { + item.status = QueueItemStatus.ready; + } + } + + setUp(() { + runner = FakeJobRunner(); + vm = MainViewModel(runner: runner); + }); + + tearDown(() => vm.dispose()); + + test('a successful job leaves the queue idle and the item completed', + () async { + await seedQueue(1); + await vm.startQueueProcessing(); + expect(runner.started, hasLength(1)); + + await runner.succeed(); + + expect(vm.queue.single.status, QueueItemStatus.completed); + expect(vm.isQueueProcessing, isFalse); + expect(vm.state, ProcessingState.completed); + }); + + test('cancelling stops the queue and allows a new run — the reported symptom', + () async { + await seedQueue(1); + await vm.startQueueProcessing(); + expect(vm.isProcessing, isTrue); + + await vm.cancelProcessing(); + + expect(vm.queue.single.status, QueueItemStatus.cancelled, + reason: 'a cancelled item must not read as failed; the queue treats ' + 'those completely differently'); + expect(vm.isQueueProcessing, isFalse); + expect(vm.state, ProcessingState.idle, + reason: 'stuck anywhere else and Start is disabled forever'); + + // The actual complaint: the next conversion never ran. + expect(vm.canProcess, isTrue); + await vm.startQueueProcessing(); + expect(runner.started, hasLength(2), + reason: 'the second job must actually start'); + expect(vm.isProcessing, isTrue); + }); + + test('cancelling recovers even when the runner reports nothing at all', + () async { + // Exactly the shipped bug: WorkerManager.cancel() returned early with a null + // process and emitted no completion, leaving `_state` latched at + // `cancelling` — which is neither `idle` nor in `canCancel`, so Start and + // Cancel were both disabled with no way back. + runner.emitNothingOnCancel = true; + await seedQueue(1); + await vm.startQueueProcessing(); + + await vm.cancelProcessing(); + + expect(vm.state, ProcessingState.idle, + reason: 'a missing completion must cost a mislabelled item, not a dead ' + 'window'); + expect(vm.isQueueProcessing, isFalse); + expect(vm.canProcess, isTrue); + }); + + test('no queue item is left stranded in `processing`', () async { + // `processing` is neither canProcess nor canReprocess, so an item left in it + // is skipped by startQueueProcessing's reset loop and can never run again. + runner.emitNothingOnCancel = true; + await seedQueue(2); + await vm.startQueueProcessing(); + + await vm.cancelProcessing(); + + expect( + vm.queue.where((q) => q.status == QueueItemStatus.processing), + isEmpty, + reason: 'an item stuck at processing poisons the queue permanently', + ); + expect(vm.queueReadyCount, greaterThan(0)); + }); + + test('an unattributable completion stands the queue down rather than latching', + () async { + await seedQueue(1); + await vm.startQueueProcessing(); + + // Two completions for one job: the second arrives with the index already + // reset. It used to be dropped silently, stranding both latches. + await runner.succeed(); + await runner.complete(const CompletionResult( + success: false, + errorMessage: 'late straggler', + )); + + expect(vm.isQueueProcessing, isFalse); + expect(vm.state.isActive, isFalse, + reason: 'a late or unattributable event must never leave the UI active'); + }); + + test('a genuine failure is still recorded as failed, not swallowed', () async { + await seedQueue(1); + await vm.startQueueProcessing(); + + await runner.fail(error: 'ffmpeg exited with -22'); + + expect(vm.queue.single.status, QueueItemStatus.failed, + reason: 'a crash reported as a cancellation would hide real breakage'); + expect(vm.isQueueProcessing, isFalse); + }); + + test('a failure mid-queue continues to the next item', () async { + await seedQueue(2); + await vm.startQueueProcessing(); + + await runner.fail(); + + expect(runner.started, hasLength(2), + reason: 'cancel stops the queue, but a failure carries on'); + expect(vm.queue.first.status, QueueItemStatus.failed); + }); + + test('a runner that cannot start does not leave the queue active', () async { + runner.failToStart = true; + await seedQueue(1); + + await vm.startQueueProcessing(); + + expect(vm.isQueueProcessing, isFalse, + reason: 'startJob threw, so nothing will ever report a completion'); + expect(vm.state.isActive, isFalse); + }); +} diff --git a/app/test/worker_manager_generation_test.dart b/app/test/worker_manager_generation_test.dart index 7607831..8f0e4b2 100644 --- a/app/test/worker_manager_generation_test.dart +++ b/app/test/worker_manager_generation_test.dart @@ -40,14 +40,6 @@ void main() { reason: 'the exit handler must bail out when superseded, or it emits ' "a stale completion and calls _cleanup() on someone else's job"); - // The guard has to come before the teardown, not after it. - final guardAt = body.indexOf('if (generation != _generation) return;'); - final cleanupAt = RegExp(r'^\s*_cleanup\(\);', multiLine: true) - .firstMatch(body)! - .start; - expect(guardAt, lessThan(cleanupAt), - reason: 'the guard must precede the _cleanup() call, or the damage is ' - 'done before it is checked'); }); test('cancel does not clean up a job it did not start', () { @@ -62,14 +54,6 @@ void main() { expect(cancelBody, contains('if (generation != _generation) return;'), reason: 'cancel() waits for a real exit, so by the time it resumes the ' 'queue may have started the next job — it must not clean that up'); - // Match the call, not the word: cancel()'s own comments mention - // _cleanup(), and comparing against prose proved nothing. - final guardAt = cancelBody.indexOf('if (generation != _generation) return;'); - final cleanupAt = RegExp(r'^\s*_cleanup\(\);', multiLine: true) - .firstMatch(cancelBody)! - .start; - expect(guardAt, lessThan(cleanupAt), - reason: 'the guard must precede the _cleanup() call'); }); test('the exit handler delegates the decision, passing the cancel flag', () { diff --git a/docs/JOB_LIFECYCLE.md b/docs/JOB_LIFECYCLE.md index 343e2e9..1060b24 100644 --- a/docs/JOB_LIFECYCLE.md +++ b/docs/JOB_LIFECYCLE.md @@ -126,16 +126,18 @@ on SIGTERM without unwinding. --- -## 5. Why cancellation is inferred, not reported +## 5. How cancellation is recognised There is no "cancelled" message in the protocol. The worker reports `complete(false)` for a cancellation and for a genuine failure alike, and `_handleStdoutLine` builds both into a `CompletionResult` with no `cancelled` flag — so it defaults to `false` (`worker_manager.dart:312`). -The app therefore has to *infer* cancellation from a flag it set itself -(`_cancelRequested`). That inference is the only thing separating two outcomes -the queue treats very differently: +So the app recognises a cancellation two ways, either of which suffices: +`_cancelRequested` (what it asked for) or **exit code 130** (what the worker +actually did). Relying on the flag alone was fragile — that is how bug (c) below +became unreachable. The distinction separates two outcomes the queue treats very +differently: | `result` | `_handleQueueItemCompletion` does | |---|---| @@ -173,9 +175,9 @@ job's stdout subscription. Fixed with `_generation`. `_pendingCompletion`, which is always set (§5), so it was unreachable. The queue saw a plain failure and started the next job. -### (d) The completion that never arrives — NOT fixed +### (d) The completion that never arrives — fixed in #67 -This is the structural one, and the likeliest cause of the remaining symptom. +This is the structural one, and the actual cause of the reported symptom. `cancelProcessing` sets `_state = cancelling` **before** doing anything, then calls `WorkerManager.cancel()`. But: @@ -214,19 +216,73 @@ cannot recover it either. That item can never be run again. --- -## 7. Invariants that should hold, and are not enforced - -1. Every started job produces **exactly one** `CompletionResult`. Currently - `_completionEmitted` enforces "at most one"; nothing enforces "at least one". -2. `_state.isActive` implies a live worker **or** a pending completion. Nothing - checks this, and nothing times out. -3. `_isQueueProcessing` implies `_currentProcessingIndex` addresses a - `processing` item. Violated whenever a completion is dropped. -4. No `QueueItem` remains `processing` once `_isQueueProcessing` is false. - Violated by the dropped-completion path, and unrecoverable because - `canReprocess` excludes `processing`. -5. Cancelling is always distinguishable from failing. Currently inferred from a - local flag rather than reported by the worker (§5). +## 7. Invariants, and what enforces them + +Each of these was violated by one of the bugs above. They are now enforced, and +the enforcement is exercised by `app/test/main_viewmodel_lifecycle_test.dart`, +which drives the real viewmodel against a `FakeJobRunner`. + +1. **Every started job produces exactly one `CompletionResult`.** + `_completionEmitted` gives "at most one"; `_jobInFlight` gives "at least + one" — `cancel()` reports even when there is no process to signal. +2. **`_state` never latches on an event that did not arrive.** `cancelProcessing` + reconciles to `idle` if nothing retired the `cancelling` latch, and logs that + it did so. +3. **An unattributable completion stands the queue down** rather than being + dropped. `_handleQueueItemCompletion` calls `_stopProcessing()` instead of + returning early. +4. **No `QueueItem` is left `processing`.** `_stopProcessing()` resets any it + finds, and every path that ends processing goes through it — so the rescue + cannot be forgotten on one of them. +5. **Cancelling is distinguishable from failing**, and is *observed* rather than + inferred: `WorkerManager.completionFor` treats `cancelRequested` **or** + `exitCode == 130` (what the worker actually returns) as cancelled. + +### Testability + +`MainViewModel` takes a `JobRunner` (`app/lib/services/job_runner.dart`), +defaulting to `WorkerManager`. Before that seam existed the queue state machine +could not be driven from a test at all, which is why three fixes for the same +hang shipped unverified — two of them "verified" by tests that only read source +text. Prefer behavioural tests here; source-scanning assertions have twice +produced false failures during ordinary refactoring. + +### (e) The stale progress file — the actual cause, fixed in #67 + +The one that produced "the UI never updates". Everything above is real, but none +of it was this. + +The worker polls `${TMPDIR}/vb_progress_${job.id}` for ffmpeg's progress, and +**`job.id` is the queue item's id** — identical every time that item is re-run. +ffmpeg writes `progress=end` as it terminates, so a cancelled run leaves one +behind. The next run's loop polls immediately, before its own ffmpeg has opened +and truncated the file, reads the previous run's tail, concludes the encode has +already finished, and **breaks out of the progress loop on its first +iteration** — then blocks forever in `decoder.wait()` while the pipeline encodes +at full speed behind it. + +Confirmed by sampling the stuck worker: `__wait4` under `execute`, 0% CPU, while +vspipe sat at 440% and both ffmpegs ran. No progress was ever reported and the +job never completed. + +Why it took four attempts to find: + +- **The app was innocent throughout.** It never received a progress event, + because none was ever sent. Every app-side fix was for a symptom. +- **It only follows a cancel**, because only a re-run of the same queue item + reuses the id. +- **It is a race** between the first poll and ffmpeg's truncate, so it came and + went. Adding `debugPrint` calls shifted the timing enough to hide it — which + is why one traced build "seemed to work" with functionally identical code. +- **No test could reproduce it**: every test generated a fresh job id, so the + file never pre-existed. Even seeding one deliberately does not fail against + the broken worker locally, because ffmpeg truncates a local file almost + instantly. The window is wide over a network share, which is where it showed. + +Fixed two ways: the file is deleted before the pipeline starts, and +`progress_end_is_ours()` refuses to believe a `progress=end` seen before this run +has reported a frame. The second is what `test_94` pins, because it is the half +that can be tested deterministically. ## 8. Preview generation diff --git a/worker/src/pipeline_executor.rs b/worker/src/pipeline_executor.rs index ed20dfc..e0f63cc 100644 --- a/worker/src/pipeline_executor.rs +++ b/worker/src/pipeline_executor.rs @@ -29,6 +29,29 @@ fn is_autoload_skip_line(line: &str) -> bool { || line.contains("forget to install a plugin") } +/// Whether a `progress=end` line belongs to the run currently executing. +/// +/// The progress file is named `vb_progress_{job.id}`, and `job.id` is the queue +/// item's id — identical every time that item is re-run. ffmpeg writes +/// `progress=end` as it terminates, so a cancelled run leaves one behind, and +/// the next run's loop polls before its own ffmpeg has opened and truncated the +/// file. +/// +/// Believing that line ends the progress loop on its first iteration. The worker +/// then blocks in `decoder.wait()` while the pipeline encodes at full speed +/// behind it: no progress is ever reported and the job never completes, so the +/// app sits on "processing" with a spinner. Four fixes were made in the app for +/// that symptom before the cause was found here. +/// +/// A `progress=end` before this run has reported a single frame cannot be ours. +/// The file is also deleted before the pipeline starts; this is the belt to that +/// pair of braces, and the part that can be tested deterministically — the file +/// race depends on how fast ffmpeg opens its output, which is why the bug showed +/// up against a network share and never locally. +fn progress_end_is_ours(current_frame: i32) -> bool { + current_frame > 0 +} + /// Format an exit status for error messages, including signal info on Unix. fn format_exit_status(status: &std::process::ExitStatus) -> String { if let Some(code) = status.code() { @@ -301,6 +324,25 @@ impl PipelineExecutor { // Build FFmpeg arguments, using a temp file for progress to avoid // Windows pipe buffering which delays progress by ~10K frames. let progress_file = std::env::temp_dir().join(format!("vb_progress_{}", job.id)); + + // Delete any file left over from a previous run of this job BEFORE the + // pipeline starts. + // + // The name is derived from job.id, which is the queue item's id and so is + // identical every time that item is re-run. ffmpeg writes `progress=end` + // as it terminates, so a cancelled run leaves one behind — and the next + // run's progress loop polls immediately, before its own ffmpeg has opened + // and truncated the file. It reads the previous run's tail, concludes the + // encode has already finished, breaks out of the loop on its first + // iteration, and blocks forever in decoder.wait() while the pipeline + // encodes at full speed behind it. No progress is ever reported and the + // job never completes: the app sits on "processing" with a spinner. + // + // Whether that happens is a race between our first poll and ffmpeg's + // truncate, which is why it came and went and never reproduced under + // test — every test used a fresh job id, so the file never pre-existed. + let _ = fs::remove_file(&progress_file); + let existing_comment = self.probe_comment(&job.input_path); let ffmpeg_args = self.build_ffmpeg_args(job, &progress_file, input_sar.as_deref(), existing_comment.as_deref()); @@ -439,7 +481,7 @@ impl PipelineExecutor { } } } else if line.starts_with("progress=end") { - ffmpeg_done = true; + ffmpeg_done = progress_end_is_ours(current_frame); } } } @@ -1246,6 +1288,40 @@ mod tests { use crate::models::{AudioCodec, AudioQuality, EncodingSettings, QTGMCParameters, VideoCodec}; use uuid::Uuid; + /// A leftover `progress=end` must not end a run that has produced no frames. + /// + /// This is the bug that produced "the UI sits on processing with a spinner + /// and never updates". The progress file is keyed on job.id, which is the + /// queue item's id and so is reused on every re-run of that item; a + /// cancelled run leaves `progress=end` behind, and the next run read it + /// before its own ffmpeg had truncated the file, broke out of the progress + /// loop immediately, and blocked in decoder.wait() forever. + /// + /// It is tested here rather than end-to-end because the failure is a race + /// between the first poll and ffmpeg's truncate: it reproduces reliably over + /// a network share and essentially never against a local file, so an + /// integration test passes with or without the fix. (Verified: the heavy + /// stale-progress tests pass against the broken worker too.) + #[test] + fn test_94_stale_progress_end_is_not_ours() { + assert!( + !progress_end_is_ours(0), + "a progress=end seen before this run reported any frame is a \ + leftover from the previous run; trusting it ends the loop on the \ + first iteration and hangs the job" + ); + } + + #[test] + fn test_95_progress_end_after_frames_ends_the_run() { + assert!( + progress_end_is_ours(1), + "once this run has reported frames, progress=end is genuinely ours \ + and must end the loop — otherwise the job never finishes" + ); + assert!(progress_end_is_ours(150_000)); + } + #[test] fn test_preview_window_centered() { // Mid-clip: window is symmetric, target sits at `radius` within it.