diff --git a/lib/src/cancelable/cancelable.dart b/lib/src/cancelable/cancelable.dart index 65a5c33..ccb2098 100644 --- a/lib/src/cancelable/cancelable.dart +++ b/lib/src/cancelable/cancelable.dart @@ -17,7 +17,11 @@ class Cancelable implements Future { factory Cancelable.fromFuture(Future future) { final completer = Completer(); future.then( - (value) => completer.complete(value), + (value) { + if (!completer.isCompleted) { + completer.complete(value); + } + }, onError: (Object e, StackTrace s) { _completeError(completer: completer, error: e, stackTrace: s); }, @@ -38,6 +42,7 @@ class Cancelable implements Future { StackTrace? stackTrace, FutureOr Function(Object error)? onError, }) { + if (completer.isCompleted) return; if (onError != null) { completer.complete(onError(error)); } else { @@ -56,6 +61,7 @@ class Cancelable implements Future { [FutureOr Function(Object error)? onError]) { final resultCompleter = Completer(); _completer.future.then((value) { + if (resultCompleter.isCompleted) return; try { resultCompleter.complete(onValue(value)); } catch (error) { @@ -84,7 +90,9 @@ class Cancelable implements Future { ) { final resultCompleter = Completer>(); Future.wait(cancelables).then((value) { - resultCompleter.complete(value); + if (!resultCompleter.isCompleted) { + resultCompleter.complete(value); + } }, onError: (Object error) { _completeError( completer: resultCompleter, diff --git a/lib/src/scheduling/executor.dart b/lib/src/scheduling/executor.dart index 4049e4c..9af2551 100644 --- a/lib/src/scheduling/executor.dart +++ b/lib/src/scheduling/executor.dart @@ -175,7 +175,10 @@ class _Executor extends Mixinable<_Executor> with _ExecutorLogger { await _initializeWorkers(); final poolSize = _pool.length; final queueSize = _queue.length; - for (int i = 0; i <= min(poolSize, queueSize); i++) { + // Schedule one drain per queued task, capped by pool size. Must be a + // strict `<`: with an empty queue min(...) == 0, and `<=` would still + // run one iteration and call _schedule() with nothing to dequeue. + for (int i = 0; i < min(poolSize, queueSize); i++) { _schedule(); } } @@ -203,6 +206,14 @@ class _Executor extends Mixinable<_Executor> with _ExecutorLogger { _ensureWorkersInitialized(); return; } + // Guard against an empty queue: _schedule() is called speculatively (on + // init warm-up and at the tail of every drain via whenComplete), so there + // may be a free worker but no task. Without this, removeFirst() throws + // StateError on the empty PriorityQueue. + if (_queue.isEmpty) { + if (_dynamicSpawning) availableWorker.kill(); + return; + } final task = _queue.removeFirst(); diff --git a/test/cancelable_from_future_race_test.dart b/test/cancelable_from_future_race_test.dart new file mode 100644 index 0000000..c369935 --- /dev/null +++ b/test/cancelable_from_future_race_test.dart @@ -0,0 +1,118 @@ +import 'dart:async'; + +import 'package:test/test.dart'; +import 'package:worker_manager/worker_manager.dart'; + +void main() { + group('Cancelable.fromFuture race condition - should not throw', () { + test('cancel after future already completed does not throw', () async { + final completer = Completer(); + final cancelable = Cancelable.fromFuture(completer.future); + + // Complete the future + completer.complete(42); + + // Allow the microtask (.then callback) to fire + await Future.delayed(Duration.zero); + + // Now cancel — should silently ignore since completer is already completed + expect( + () => cancelable.cancel(), + returnsNormally, + ); + }); + + test( + 'cancel before future completes does not cause StateError', + () async { + final futureCompleter = Completer(); + final caughtErrors = []; + + await runZonedGuarded( + () async { + final cancelable = Cancelable.fromFuture(futureCompleter.future); + + // Handle the CanceledError on the future so it doesn't leak + cancelable.future.catchError((_) => -1); + + // Cancel first — completes the internal completer with CanceledError + cancelable.cancel(); + + // Now the future resolves — .then fires, but should silently ignore + futureCompleter.complete(42); + + // Allow microtasks to process + await Future.delayed(Duration.zero); + }, + (error, stack) { + caughtErrors.add(error); + }, + ); + + // No StateError should have been caught — only CanceledError is acceptable + expect(caughtErrors.whereType(), isEmpty); + }); + + test('_completeError guards against already-completed completer', () async { + final completer = Completer(); + + // Handle the future error so it doesn't leak as unhandled + completer.future.catchError((_) => -1); + + completer.completeError(CanceledError()); // first completion + + // The second attempt via _completeError should be silently ignored + // (we can't directly call _completeError, but cancel() triggers it) + final cancelable = Cancelable( + completer: completer, + onCancel: () { + // This simulates what fromFuture's onCancel does + if (!completer.isCompleted) { + completer.completeError(CanceledError()); + } + }, + ); + + expect(() => cancelable.cancel(), returnsNormally); + }); + + test('mergeAll cancel after futures complete does not throw', () async { + final c1 = Completer(); + final c2 = Completer(); + + final cancelable1 = Cancelable.fromFuture(c1.future); + final cancelable2 = Cancelable.fromFuture(c2.future); + + final merged = Cancelable.mergeAll([cancelable1, cancelable2]); + + c1.complete(1); + c2.complete(2); + await Future.delayed(Duration.zero); + + // Cancel after all futures resolved — should silently ignore + expect( + () => merged.cancel(), + returnsNormally, + ); + }); + + test('cancelable still resolves with value when not cancelled', () async { + final completer = Completer(); + final cancelable = Cancelable.fromFuture(completer.future); + + completer.complete(42); + + final result = await cancelable; + expect(result, 42); + }); + + test('cancelable resolves with CanceledError when cancelled before completion', () async { + final completer = Completer(); + final cancelable = Cancelable.fromFuture(completer.future); + + cancelable.cancel(); + + expect(() => cancelable.future, throwsA(isA())); + }); + }); +} diff --git a/test/init_with_empty_queue_test.dart b/test/init_with_empty_queue_test.dart new file mode 100644 index 0000000..c58e6b9 --- /dev/null +++ b/test/init_with_empty_queue_test.dart @@ -0,0 +1,34 @@ +// ignore_for_file: invalid_use_of_visible_for_testing_member +import 'package:test/test.dart'; +import 'package:worker_manager/worker_manager.dart'; + +/// Regression test for the crash introduced by PR #126, where init() with the +/// default (non-dynamic) spawning threw `StateError: No element` because the +/// warm-up loop scheduled once against an empty queue and _schedule() had lost +/// its empty-queue guard. +void main() { + setUp(() async => await workerManager.dispose()); + tearDown(() async => await workerManager.dispose()); + + test('init() with default (non-dynamic) spawning warms the pool without ' + 'crashing on an empty queue', () async { + // Must not throw StateError from an unguarded _queue.removeFirst(). + await workerManager.init(); + + expect(workerManager.pool.isNotEmpty, isTrue); + expect( + workerManager.pool.every((w) => w.initialized), + isTrue, + reason: 'non-dynamic init() eagerly initializes all workers', + ); + }); + + test('init() then a single task still resolves (queue drains to empty ' + 'without crashing the tail _schedule())', () async { + await workerManager.init(); + + final int result = await workerManager.execute(() => 21 * 2); + + expect(result, 42); + }); +}