Can use static function here to avoid holding a reference to $this.

This leaks the amp/future directly, which is somewhat tricky to use. Ideally, we would just allow getting the value (or waiting for it) instead of leaking the internal implementation.

and others in this file?

Didn't we already do this on line 347?

It'd be great if we could factor out the multiple layers of futures.

I feel like we can take advantage of some amp stuff to make this much more readable.

See: Future::iterate(), for example. I find this entire loop quite confusing. We are iterating over $futures, while also maintaining a copy in $executions. I'm not 100% sure any of this is actually doing anything in parallel.

For example, we set $i = 0, then wait for something to complete, then filter out the completed future from $executions, then reset $futures, then increment i, then wait for all executions to complete, then start all over again.

Like, I'm not sure what the intent is here; some comments would be helpful to explain why you are doing what you are doing. To be frank, it doesn't seem to make any sense at all.

First, it appears that you want to set a processing limit, but all these fibers/processes are already running. You can't set the parallel processing limit this late in the code. Instead, you should set it in src/Symfony/Component/Messenger/ParallelMessageBus.php:38 when you set up the worker pool. That would be where you set the limits by defining how many threads / fibers it will run concurrently. Doing this here doesn't do anything.

Secondly, the entire code can be written differently to take advantage of amp's parallelism.

$futures = self::$futures; // take a copy of futures we care about
self::$futures = []; // any other fiber shouldn't watch these fibers

// loop over all tasks in the background.
async(
  function () use ($futures)
  {
    // iterate over futures in completion order
    foreach(Future::iterate($futures) as $idx => $future) {
      try {
        $val = $future->await();
        // do stuff with val
      } catch (\Throwable $ex) {
        // handle error of $futures[$idx] here.
      }
    }
  }
)

Edit to add:

See Future::map() to attach a callback on completion. Here's a queue handler I wrote awhile ago using amphp that could handle ~5k jobs per second: https://github.com/bottledcode/durable-php/blob/main/src/Processor.php#L150-L185

thanks for your feedback.
Instead of calling awaitFirst, I could implement the internal logic of this function in the worker code. This way, I would have access to the index of the Future, which would make it easier to unset and sort what is done or still pending.
But it's wrap in an Executionclass so it would be something like

foreach(Future::iterate($futuresReceived) as $idx => $future) {
            try {
                $execution = $future->await();
                $execution->getFuture()->await();
                $envelope = $execution->getFuture()->await();

                unset($futuresReceived[$idx]);

                $this->preAck($envelope, $transportName, false, null);
            } catch (\Throwable $ex) {
            }
            ```

Regarding the action, when using the parallel message bus the messages seem to be processed concurrently rather than in order. I'll add a small screenshot to show the result compared to a "normal" bus

Regarding the action, when using the parallel message bus the messages seem to be processed concurrently rather than in order. I'll add a small screenshot to show the result compared to a "normal" bus

Yes, the bus is processing concurrently, but you were not checking them concurrently nor setting limits on concurrency.

$parallelProcessLimit seems unused