fix: resolve race-condition deadlock in parallelForEach using package:pool

[sigurdm]

Description

Fixes a race-condition deadlock in StreamExtensions.parallelForEach that caused background tasks such as check-datastore-integrity and synchronize-exported-api to hang indefinitely.

Root Cause of the Deadlock

The Notifier helper class acts as an event broadcaster rather than a counting semaphore:

final class Notifier {
  var _completer = Completer<void>();

  void notify() {
    if (!_completer.isCompleted) {
      _completer.complete();
    }
  }

  Future<void> get wait {
    if (_completer.isCompleted) {
      _completer = Completer();
    }
    return _completer.future;
  }
}

When multiple tasks complete close together:

1. Notifier.notify() marks its internal _completer as completed.
2. Any subsequent notify() calls occurring before wait is accessed are dropped (!_completer.isCompleted is false).
3. When await itemDone.wait is subsequently called, if (_completer.isCompleted) resets _completer = Completer() and returns an uncompleted Future from the new completer, waiting for a subsequent notification instead of resolving immediately.
4. When parallelForEach completes its await for stream iteration and enters the finally block:

   while (running > 0) {
     await itemDone.wait;
   }

   If remaining in-flight tasks finish while wait is accessed or during a microtask turn, running becomes 0, but wait returns a brand new unresolved Completer. Because no tasks remain running to call notify(), parallelForEach deadlocks indefinitely.

Minimal Reproduction Example

import 'dart:async';
import 'package:pub_dev/shared/parallel_foreach.dart';

void main() async {
  // A stream that yields items with micro-delays (e.g. paginated queries).
  Stream<int> stream() async* {
    for (var i = 0; i < 100; i++) {
      yield i;
    }
  }

  // With fast async tasks, remaining in-flight tasks complete while `wait`
  // is evaluated in `finally`, creating an unresolved Completer that never resolves.
  await stream().parallelForEach(4, (item) async {
    await Stream.fromIterable([item]).toList();
  }); // <-- Deadlocks here indefinitely
}

Fix

• Refactors parallelForEach to use Pool from package:pool (already in pubspec.yaml).
• await pool.request() cleanly enforces backpressure on the stream when maxParallel tasks are in-flight.
• resource.release() immediately grants execution to the next waiting request in FIFO order.
• await pool.close() safely waits for all in-flight tasks to complete in the finally block.
• Adds stress tests in parallel_foreach_test.dart to verify no race conditions or deadlocks occur under high-throughput streaming workloads.

[jonasfj]

I do not believe the explanations given here.

I could be wrong, but when reading:

while (running > 0) {
  await itemDone.wait;
}

yes, the Completer and by implication the future is created inside .wait. BUT nothing can run between running > 0 and itemDone.wait. Other microtasks cannot run and call .notify inbetween statements as claimed in the PR description:

If remaining in-flight tasks finish while wait is accessed or during a microtask turn,
running becomes 0, but wait returns a brand new unresolved Completer.

We have code like:

// When [each] is done, we decrement [running] and notify
running -= 1;
itemDone.notify();

The decrement cannot happen after running > 0 and before itemDone.wait.

[sigurdm]

I think you are right - this won't help :/