From 9009525b7b53ed28c7a4a743685a48f28d27a0fe Mon Sep 17 00:00:00 2001 From: Stuart Cameron Date: Sun, 9 Aug 2026 11:13:04 +1000 Subject: [PATCH 1/4] fix(queue): guarantee a completion, and make the state machine testable Fourth attempt at the cancellation hang, and the first with a test that drives the thing that was actually broken. The hang is structural. 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 silently at its guard, and the UI spins with Start AND Cancel disabled, because `cancelling` is neither `idle` nor in `canCancel`. There is no way back short of restarting the app. The event genuinely does not always arrive. `cancel()` returned early when `_process` was null and reported nothing, while `cancelProcessing` had already latched `_state = cancelling`. `_process` is null more often than it looks -- notably during `preparingJob`, where the Cancel button is live before the process exists. Fixes, all now enforced rather than assumed: - `_jobInFlight` gives "at least one completion per started job"; only "at most one" was enforced before. cancel() reports even with nothing to signal. - Cancellation is observed, not merely inferred: exit code 130 counts alongside `_cancelRequested`. Depending on the flag alone is how the previous fix became unreachable. - An unattributable completion stands the queue down instead of being dropped. - `_stopProcessing()` is the single stand-down path, and it rescues items stranded in `processing` -- a status that is neither canProcess nor canReprocess, so such an item could never run again. - `cancelProcessing` reconciles to idle if nothing retired the latch, and logs it. A missed event should cost a mislabelled item, not a dead window. The reason three previous fixes shipped unverified is that MainViewModel constructed its own WorkerManager, so the queue state machine could not be driven from a test at all. It now takes a JobRunner (defaulting to WorkerManager), and main_viewmodel_lifecycle_test drives the real viewmodel against a fake: cancel then start again, cancel with no completion emitted, a stale completion, a stranded item, a failure mid-queue. Two of the eight fail against the previous code -- the two that describe the shipped bug. Also retires the positional source-scanning assertions added earlier. They anchored on "the first _cleanup()", which the new early-return branch displaced, and produced false failures with the behaviour perfectly correct. Twice now those tests have cost time without catching anything; behaviour covers them. --- app/lib/services/job_runner.dart | 48 ++++++ app/lib/services/worker_manager.dart | 53 +++++- app/lib/viewmodels/main_viewmodel.dart | 64 ++++++- app/test/cancel_shutdown_grace_test.dart | 7 +- app/test/fakes/fake_job_runner.dart | 93 ++++++++++ app/test/main_viewmodel_lifecycle_test.dart | 171 +++++++++++++++++++ app/test/worker_manager_generation_test.dart | 16 -- docs/JOB_LIFECYCLE.md | 57 ++++--- 8 files changed, 461 insertions(+), 48 deletions(-) create mode 100644 app/lib/services/job_runner.dart create mode 100644 app/test/fakes/fake_job_runner.dart create mode 100644 app/test/main_viewmodel_lifecycle_test.dart 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/worker_manager.dart b/app/lib/services/worker_manager.dart index 5508648..b2323bc 100644 --- a/app/lib/services/worker_manager.dart +++ b/app/lib/services/worker_manager.dart @@ -4,12 +4,13 @@ import 'dart:io'; 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 +52,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 +91,7 @@ class WorkerManager { _completionEmitted = false; _cancelRequested = false; + _jobInFlight = true; final generation = ++_generation; final toolLocator = ToolLocator.instance; @@ -220,17 +236,39 @@ 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 +348,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, @@ -333,6 +378,7 @@ class WorkerManager { void _emitCompletion(CompletionResult result) { if (_completionEmitted || _completionController.isClosed) return; _completionEmitted = true; + _jobInFlight = false; _completionController.add(result); } @@ -396,6 +442,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..8918749 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,21 @@ class MainViewModel extends ChangeNotifier { } else { await _workerManager.cancel(); } + + // 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..d93b08b 100644 --- a/app/test/cancel_shutdown_grace_test.dart +++ b/app/test/cancel_shutdown_grace_test.dart @@ -89,7 +89,12 @@ void main() { .readAsStringSync(); 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..38c1d51 --- /dev/null +++ b/app/test/fakes/fake_job_runner.dart @@ -0,0 +1,93 @@ +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); + } + + 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/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..39d708d 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,36 @@ 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. ## 8. Preview generation From d90a5d90a811a3f5d107b77bbf4da2bddbcbfa43 Mon Sep 17 00:00:00 2001 From: Stuart Cameron Date: Sun, 9 Aug 2026 11:29:17 +1000 Subject: [PATCH 2/4] test(worker): make WorkerManager drivable against a real worker Adds a VAPOURBOX_WORKER override to ToolLocator._resolveWorker, mirroring the existing VAPOURBOX_DEPS_DIR one. Under `flutter test` the resolved executable is the test runner, so neither the production nor the dev lookup could find the worker binary -- which left WorkerManager, and with it the whole cancel/restart flow, with no way to be tested against a real worker at all. Same structural gap as the viewmodel had, one layer down. Two heavy tests with that seam, both currently passing: - progress still flows after a cancel and a restart - a restart that lands while the cancel is still in flight still reports, which models what the UI actually does (progress_panel.dart:440 does not await cancelProcessing, so a user can restart mid-cancel) Neither reproduces the reported symptom. Recording that as evidence rather than as a fix: the viewmodel relays progress correctly against a fake runner, and WorkerManager delivers it across both restart paths against the real worker, so the fault is somewhere neither layer models. They stand as regression cover for the parts now known good. --- app/lib/services/tool_locator.dart | 10 ++ app/test/fakes/fake_job_runner.dart | 2 + ...tegration_worker_manager_restart_test.dart | 162 ++++++++++++++++++ 3 files changed, 174 insertions(+) create mode 100644 app/test/integration_worker_manager_restart_test.dart 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/test/fakes/fake_job_runner.dart b/app/test/fakes/fake_job_runner.dart index 38c1d51..0c68c97 100644 --- a/app/test/fakes/fake_job_runner.dart +++ b/app/test/fakes/fake_job_runner.dart @@ -85,6 +85,8 @@ class FakeJobRunner implements JobRunner { 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)); 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))); +} From 53064e69a7f76fce3e6410aeb345e9283e39c362 Mon Sep 17 00:00:00 2001 From: Stuart Cameron Date: Sun, 9 Aug 2026 11:57:08 +1000 Subject: [PATCH 3/4] fix(worker): ignore a previous run's progress file The actual cause of "the UI sits on processing and never updates". Four fixes were made in the app for this symptom; the app was innocent throughout, because it never received a progress event -- none was ever sent. The worker polls ${TMPDIR}/vb_progress_${job.id}, and job.id is the queue item's id, so it is 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 before its own ffmpeg has opened and truncated the file, reads the previous run's tail, concludes the encode is finished, breaks out of the progress loop on its first iteration, and then blocks in decoder.wait() forever 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. Fixed twice over: the file is deleted before the pipeline starts, and progress_end_is_ours() refuses a `progress=end` seen before this run has reported a frame. On the tests: the heavy integration tests seed a stale file and assert the job still completes and still reports progress, but they pass against the BROKEN worker too -- locally ffmpeg truncates the file almost instantly, so the race window barely exists. It is wide over a network share, which is where this surfaced. They are smoke cover, not the regression guard. The guard is test_94, on the extracted decision, which fails against the previous behaviour. Same lesson as completionFor: when the bug is a decision, extract the decision and test it, rather than hoping an end-to-end test happens to hit the timing. Also removes the temporary lifecycle tracing added to find this. --- app/lib/services/worker_manager.dart | 8 +- app/lib/viewmodels/main_viewmodel.dart | 8 + .../integration_stale_progress_file_test.dart | 137 ++++++++++++++++++ docs/JOB_LIFECYCLE.md | 37 +++++ worker/src/pipeline_executor.rs | 78 +++++++++- 5 files changed, 266 insertions(+), 2 deletions(-) create mode 100644 app/test/integration_stale_progress_file_test.dart diff --git a/app/lib/services/worker_manager.dart b/app/lib/services/worker_manager.dart index b2323bc..cf8d889 100644 --- a/app/lib/services/worker_manager.dart +++ b/app/lib/services/worker_manager.dart @@ -2,6 +2,8 @@ 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'; @@ -115,6 +117,7 @@ class WorkerManager implements JobRunner { workingDirectory: File(workerPath).parent.path, ); + // Listen to stdout for JSON messages _stdoutSubscription = _process!.stdout .transform(utf8.decoder) @@ -247,6 +250,7 @@ class WorkerManager implements JobRunner { 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 @@ -376,7 +380,9 @@ class WorkerManager implements JobRunner { /// 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); diff --git a/app/lib/viewmodels/main_viewmodel.dart b/app/lib/viewmodels/main_viewmodel.dart index 8918749..ba906a7 100644 --- a/app/lib/viewmodels/main_viewmodel.dart +++ b/app/lib/viewmodels/main_viewmodel.dart @@ -1287,6 +1287,14 @@ class MainViewModel extends ChangeNotifier { 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 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/docs/JOB_LIFECYCLE.md b/docs/JOB_LIFECYCLE.md index 39d708d..1060b24 100644 --- a/docs/JOB_LIFECYCLE.md +++ b/docs/JOB_LIFECYCLE.md @@ -247,6 +247,43 @@ hang shipped unverified — two of them "verified" by tests that only read sourc 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 A separate lifecycle with different rules, worth knowing because it shares the 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. From 9cfca4ea036f029e9e191f8902adaa0536bc46f1 Mon Sep 17 00:00:00 2001 From: Stuart Cameron Date: Sun, 9 Aug 2026 14:39:13 +1000 Subject: [PATCH 4/4] fix(test): normalise line endings in the remaining source scan Third CI break from the same cause. git checks worker_manager.dart out CRLF on Windows, so indexOf('\n }\n') returned -1 and substring threw a RangeError. The sibling file was normalised two commits ago; this one was missed. Adds a note at the top of the file, because the failure mode is opaque -- it has presented once as a RangeError and twice as a nonsensical 'expected a value less than N' -- and because the right answer is to prefer behavioural assertions over reading source at all. These remain only because what they guard is a code shape rather than an observable. --- app/test/cancel_shutdown_grace_test.dart | 15 ++++++++++++--- 1 file changed, 12 insertions(+), 3 deletions(-) diff --git a/app/test/cancel_shutdown_grace_test.dart b/app/test/cancel_shutdown_grace_test.dart index d93b08b..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,7 +94,8 @@ 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)); // Read the whole method, not up to the first `_cleanup()`. cancel() now