From b1cf6d506fa47e0c5840ba4c2ee3ea662d4ec640 Mon Sep 17 00:00:00 2001 From: RWRZ Date: Fri, 10 Apr 2026 12:40:18 -0300 Subject: [PATCH 1/2] fix: guard Cancelable against double-completion race condition Cancelable.fromFuture, _completeError, thenNext, and mergeAll could all throw "Bad state: Future already completed" when cancel() raced with the wrapped Future resolving. Added isCompleted guards before every completer.complete/completeError call. Co-Authored-By: Claude Opus 4.6 (1M context) --- lib/src/cancelable/cancelable.dart | 12 ++- test/cancelable_from_future_race_test.dart | 118 +++++++++++++++++++++ 2 files changed, 128 insertions(+), 2 deletions(-) create mode 100644 test/cancelable_from_future_race_test.dart 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/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())); + }); + }); +} From 1a50e54f8278b0f9aa7ca962e1d9e50336f2c960 Mon Sep 17 00:00:00 2001 From: RWRZ Date: Mon, 6 Jul 2026 13:24:27 -0300 Subject: [PATCH 2/2] fix(executor): guard _schedule() against empty queue + off-by-one warm-up PR #126 (refactor_issue_124) regressed _schedule() in two ways that crash init() on the default dynamicSpawning: false path, bricking app startup: - It deleted the `if (_queue.isEmpty) { ... return; }` guard right before `_queue.removeFirst()`, so removeFirst() throws `StateError: No element` on the empty PriorityQueue. - `_ensureWorkersInitialized()` warms up with `i <= min(poolSize, queueSize)`. At init the queue is empty, so min(N, 0) == 0, but the `<=` bound still runs one iteration and calls _schedule() with nothing to dequeue -> crash. _schedule() is also called speculatively at the tail of every drain (a just-freed worker + empty queue via whenComplete), so the missing guard bit beyond init too. The PR's own test only covered dynamicSpawning: true, which skips both paths. Restore the empty-queue guard and make the warm-up loop bound strict (`<`). Add test/init_with_empty_queue_test.dart covering the non-dynamic init() path. Co-Authored-By: Claude Opus 4.8 (1M context) Claude-Session: https://claude.ai/code/session_01Hz16YUNUjSgMq1zZfPFdFJ --- lib/src/scheduling/executor.dart | 13 ++++++++++- test/init_with_empty_queue_test.dart | 34 ++++++++++++++++++++++++++++ 2 files changed, 46 insertions(+), 1 deletion(-) create mode 100644 test/init_with_empty_queue_test.dart 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/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); + }); +}