So, if there are 100 delayed messages, and only 5 delayed messages that actually are available now, wouldn't this end up cycling through the 100 delayed messages?

@ragboyjr no $queuedMessageCount will be 5 then and so it only pop 5 messages.

Ah, understood.

Your technique for handling the race condition seems nice and simple, but one thing that concerns me would be loss of messages between the time that we pop it, then your redis instance has some failure, and then it fails when we try to re-add it (in the event of that race condition.

Part of me wonders if maybe, doing something where we pass in a LUA script to where it pops and pushes into the stream in one transaction would be preferable since the whole transaction would fail instead of part way through. What do you think?

@ragboyjr As redis has only features to move a message from list to another list and not from a list to a stream. we would need to create a transaction then. never used that but I think it should be something like this:

    $this->connection->multi();
    foreach ($this->connection->zpopmin($this->queue, $queuedMessageCount) as $queuedMessage => $time) {
                $queuedMessage = json_encode($queuedMessage, true);
                // if a futured placed message is actually popped because of a race condition with
                // another running message consumer, the message is readded to the queue by add function
                // else its just added stream and will be available for all stream consumers
                $this->add($queuedMessage['body'], $queuedMessage['headers'], (time() - $time) * 1000);
    }

    $this->connection->exec();

we only need to start the transaction if there are really queued messages so its not blocking if there are none.

Y, I think a redis transaction would work as well, would like to see if that could be something we could cover in an integration test of adding items to the queue, then having redis fail and making sure we don't lose any messages.

Transactions sadly don't work as we can't read inside a transaction.
So to only solution would be to park this messages in another list as every consumer need its own name we postfix the new list with stream + group + consumer names and so its a unique list only for this consumer to avoid a conflicts with other running consumers. If it crash the consumer need to check if he has still things in this list to push to the stream.