Adding jobs to a batch throws Redis::pipeline(): Can't activate pipeline in multi mode

[seabass86]

• Laravel Version: 8.36.2
• PHP Version: 8.0.3
• Database Driver & Version: -

Description:

From within a queued job I want to create a new batch processing job:

$batch = Bus::batch([])
            ->allowFailures()
            ->finally(function () use ($campaign) {
               ...
            })
            ->name($campaign->getBatchName())
            ->onQueue(...)
            ->dispatch();

Once I am adding jobs to this batch $batch->add(...);

I receive the exception

{
    "class": "Symfony\\Component\\ErrorHandler\\Error\\FatalError",
    "message": "Redis::pipeline(): Can't activate pipeline in multi mode!",
    "code": 0,
    "file": "/var/www/versions/current/vendor/laravel/framework/src/Illuminate/Redis/Connections/PhpRedisConnection.php:402"
}

Horizon is running on its own redis connection. Not the default on. Dispatching jobs from within jobs works fine. Just using the batch function does not.

I believe it to be a bug, as apparently the redis connector is trying to use two different redis modes both established by 1st party laravel packages.

Thank you :)

[seabass86]

So when I change the line in PhpRedisConnection.php:402 from $pipeline = $this->client()->pipeline(); to $pipeline = $this->client()->multi(); all issues are gone. The batch processing seems to switch to pipeline mode even though a multimode is already open, which seems to not be allowed. Eventually this happens at src/Illuminate/Queue/RedisQueue.php:97 public function bulk($jobs, $data = '', $queue = null)

 {
        $this->getConnection()->pipeline(function () use ($jobs, $data, $queue) {
            $this->getConnection()->transaction(function () use ($jobs, $data, $queue) {
                foreach ((array) $jobs as $job) {
                    $this->push($job, $data, $queue);
                }
            });
        });
    }

[taylorotwell]

Where do you actually call $batch->add() in your example code?

[seabass86]

The batch is created from within a queued job:

public function handle()
    {
    $batch = Bus::batch([])
            ->allowFailures()
            ->finally(function () {
                // ...
            })
            ->name($this->batchName)
            ->onQueue('mail')
            ->dispatch();
    // ...
    foreach($this->models as $model){
        $batch->add([new CreateInvitationJob($model)]);
    }
}

The redis exception is thrown at $batch->add.

[taylorotwell]

@themsaid do you have any ideas on this?

[themsaid]

@seabass86 I can't replicate the issue on PHPRedis or Predis. Adding jobs to an existing batch works fine. Haven't seen any similar reports either.

[driesvints]

@seabass86 since there's no clear way to reproduce this I'm going to close this one. If you have specific steps to reproduce feel free to open a new issue.

[KennedyTedesco]

I was able to reproduce this. Occurs when you have two Redis connections for the same server (even if the database is different), so, at some point, he tries to activate the pipeline mode (when using batch) in a connection the already started using another mode.

If you have just one connection, this never happens. I'm going to spend some time on this soon to see if I get it fixed.

Fun fact: The issue only occurs when using the phpredis extension. With predis all goes fine.

[robbinjanssen]

@KennedyTedesco did you ever found a fix/resolve this problem? :-)

[KennedyTedesco]

@robbinjanssen I don't remember what I did to deal with this, but I'm confident that updating PHP, Redis, and phpredis to the last version probably fixed my issue. I don't remember the last time I saw this issue. But, indeed, it was a real issue back then.

[Tachii]

It happens when you set after_commit = true in database.php for redis

'redis' => [
            'driver'       => 'redis',
            'connection'   => 'horizon',
            'queue'        => env(key: 'REDIS_QUEUE', default: 'default'),
            'retry_after'  => 10,
            'block_for'    => null,
            'after_commit' => false, // set it to false
        ],