Dev

lib/src/cancelable/cancelable.dart

@@ -17,7 +17,11 @@ class Cancelable<R> implements Future<R> {
   factory Cancelable.fromFuture(Future<R> future) {
     final completer = Completer<R>();
     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<R> implements Future<R> {
     StackTrace? stackTrace,
     FutureOr<T> Function(Object error)? onError,
   }) {
+    if (completer.isCompleted) return;
     if (onError != null) {
       completer.complete(onError(error));
     } else {
@@ -56,6 +61,7 @@ class Cancelable<R> implements Future<R> {
       [FutureOr<T> Function(Object error)? onError]) {
     final resultCompleter = Completer<T>();
     _completer.future.then((value) {
+      if (resultCompleter.isCompleted) return;
       try {
         resultCompleter.complete(onValue(value));
       } catch (error) {
@@ -84,7 +90,9 @@ class Cancelable<R> implements Future<R> {
   ) {
     final resultCompleter = Completer<Iterable<T>>();
     Future.wait(cancelables).then((value) {
-      resultCompleter.complete(value);
+      if (!resultCompleter.isCompleted) {
+        resultCompleter.complete(value);
+      }
     }, onError: (Object error) {
       _completeError(
         completer: resultCompleter,

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();
 

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<int>();
+      final cancelable = Cancelable<int>.fromFuture(completer.future);
+
+      // Complete the future
+      completer.complete(42);
+
+      // Allow the microtask (.then callback) to fire
+      await Future<void>.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<int>();
+      final caughtErrors = <Object>[];
+
+      await runZonedGuarded(
+        () async {
+          final cancelable = Cancelable<int>.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<void>.delayed(Duration.zero);
+        },
+        (error, stack) {
+          caughtErrors.add(error);
+        },
+      );
+
+      // No StateError should have been caught — only CanceledError is acceptable
+      expect(caughtErrors.whereType<StateError>(), isEmpty);
+    });
+
+    test('_completeError guards against already-completed completer', () async {
+      final completer = Completer<int>();
+
+      // 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<int>(
+        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<int>();
+      final c2 = Completer<int>();
+
+      final cancelable1 = Cancelable<int>.fromFuture(c1.future);
+      final cancelable2 = Cancelable<int>.fromFuture(c2.future);
+
+      final merged = Cancelable.mergeAll([cancelable1, cancelable2]);
+
+      c1.complete(1);
+      c2.complete(2);
+      await Future<void>.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<int>();
+      final cancelable = Cancelable<int>.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<int>();
+      final cancelable = Cancelable<int>.fromFuture(completer.future);
+
+      cancelable.cancel();
+
+      expect(() => cancelable.future, throwsA(isA<CanceledError>()));
+    });
+  });
+}

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