Unexpected behavior with Batch callbacks with Queue

[josephkerkhof]

Hi all 👋

I'm observing a behavior that is unexpected for me.

Let's say I'm batching some jobs that I want to go to a queue other than the default, let's call this other queue "import".

Bus::batch([
    new MyJob,
])
    ->onQueue(Queue::Import)
    ->dispatch(); // dispatches to the "import" queue

This will successfully add the batch of jobs to the "import" queue as expected.

However, if I chain a callback function in the batch, the jobs will instead go to the "default" queue.

Bus::batch([
    new MyJob,
    new MyJob,
    new MyJob,
])
    ->then(fn () => Log::info('finished MyJobs!'))
    ->onQueue(Queue::Import)
    ->dispatch(); // dispatches to the "default" queue

I've noticed this problem also occurs with the catch() callback as well. This seems like unexpected behavior since the documentation shows us that this should be possible.

I am on version 12.37.0 of the laravel/framework.

[macropay-solutions]

Can you please check in the batched table from DB the serialized options for both cases?

[macropay-solutions]

Also please check that your first job from the second example does not have the default queue defined in it.

[josephkerkhof]

Sure, I here are the serialized options for both cases. Please note that my actual queue on my local machine that I'm working with is actually called "snowflake" and not "import" as I mentioned in my first post.

For this example:

Bus::batch([
    new MyJob,
])
    ->onQueue(Queue::Snowflake)
    ->dispatch(); // dispatches to the "snowflake" queue

the serialized options look like this:

a:1:{s:5:"queue";s:9:"snowflake";}

and the horizon dashboard looks like this:

For this example:

Bus::batch([
    new MyJob,
    new MyJob,
    new MyJob,
])
    ->then(fn () => Log::info('finished MyJobs!'))
    ->onQueue(Queue::Snowflake)
    ->dispatch(); // dispatches to the "default" queue

the serialized options look like this:

a:2:{s:4:"then";a:1:{i:0;O:47:"Laravel\SerializableClosure\SerializableClosure":1:{s:12:"serializable";O:46:"Laravel\SerializableClosure\Serializers\Signed":2:{s:12:"serializable";s:174:"O:46:"Laravel\SerializableClosure\Serializers\Native":5:{s:3:"use";a:0:{}s:8:"function";s:0:"";s:5:"scope";N;s:4:"this";N;s:4:"self";s:32:"0000000000000c290000000000000000";}";s:4:"hash";s:44:"mDmbVOJnYGlSEz0gK51XzNRC5B+EzAVZSxuMLcFIjrg=";}}}s:5:"queue";s:9:"snowflake";}

and the horizon dashboard looks like this:

What is interesting about that serialized data is that it seems to correctly be identifying the snowflake queue in both cases, yet in my horizon dashboard, I see the jobs going to the wrong queue.

[macropay-solutions]

Well seeing them in horizon and actually being in the wrong queue are 2 different stories. Can you confirm this is not a display bug from horizon?

[josephkerkhof]

Yep, it's not a bug in the horizon UI. If I open a tinker shell and run these commands, this is what I see:

> Queue::size()
= 3

> Queue::getConnectionName()
= "redis"

> Queue::size('default')
= 3

> Queue::size('snowflake')
= 0

This is consistent with what shows in the web UI.

[macropay-solutions]

UPDATE. Disregard this. Look below. The problem was related to tinker.

The issue is that the queue is not set on the event

BatchDispatched class has only a batch property and

// \Illuminate\Broadcasting\BroadcastManager::queue

        $queue = null;

        if (method_exists($event, 'broadcastQueue')) {
            $queue = $event->broadcastQueue();
        } elseif (isset($event->broadcastQueue)) {
            $queue = $event->broadcastQueue;
        } elseif (isset($event->queue)) {
            $queue = $event->queue;
        }

will not see the queue.

[macropay-solutions]

What if you define a method in that class broadcastQueue that returns that snowflake?

[josephkerkhof]

Nope, that still doesn't work.

if I put a dd() in the BroadcastManager file, I'm not getting it to trip.

[josephkerkhof]

I'm looking at the \Illuminate\Events\Dispatcher::dispatch function to see if that might point me in the right direction.

[macropay-solutions]

Dead end...

[macropay-solutions]

//\Illuminate\Bus\PendingBatch::dispatch
    public function dispatch()
    {
        $repository = $this->container->make(BatchRepository::class);

        try {
            $batch = $this->store($repository);

            $batch = $batch->add($this->jobs);
//\Illuminate\Bus\Batch::add
        $this->repository->transaction(function () use ($jobs, $count) {
            $this->repository->incrementTotalJobs($this->id, $count);

            $this->queue->connection($this->options['connection'] ?? null)->bulk(
                $jobs->all(),
                $data = '',
                $this->options['queue'] ?? null
            );
        });

        return $this->fresh();

// \Illuminate\Queue\RedisQueue::bulk

This code should be the one executing. Can you hardcode your queue instead of$this->options['queue'] ?? null?

[josephkerkhof]

I think I'm getting somewhere on this. The issue has something to do with the serialization process of the $this->options on the PendingBatch (\Illuminate\Bus\DatabaseBatchRepository::store).

For some reason the callback function is corrupts the serialized data. Because it's corrupted, the deserializing function is just returning an empty array (vendor/laravel/framework/src/Illuminate/Bus/DatabaseBatchRepository.php:376), therefore wiping out the queue designation entirely.

It's worth pointing out that I've been running my Bus::batch(...) in a tinkerwell (and regular Laravel tinker) script.

I then went to look at the laravel/serializable-closure package and noticed that in \Laravel\SerializableClosure\Support\ReflectionClosure::getFileTokens:812, there's a call to file_get_contents($this->getFileName()). This is where the serialization breaks down I think. It seems the tinker scripts are not able to be read from disk. In fact, I notice that I get a warning when running this with Laravel Tinker:

   WARNING  file_get_contents(eval()'d code): Failed to open stream: No such file or directory in 
vendor/laravel/serializable-closure/src/Support/ReflectionClosure.php on line 812.

I believe this is what is corrupting the $this->options serialization for the PendingBatch.

[marius-ciclistu]

Nice

[macropay-solutions]

Is it replicating in a sandbox or prod deploy?

[josephkerkhof]

Well, I've figured out that the laravel/serializable-closure doesn't support the tinker REPL: https://github.com/laravel/serializable-closure

I think that pretty much concludes this adventure. I'm not sure if the Bus::batch(...)->then(...)->dispatch() should throw a warning or exception when in the tinker REPL...

Big thank you to @macropay-solutions for helping talk through the issue! 👏

[macropay-solutions]

Thank you. We went through this bus batch and now we understand it better. Each job will be sent on the queue as a message just like if you would dispatch it alone. The batch is just developer convenience to see in UI the batch stats. We use something similar but we implemented our own solution with increments which is way simpler than the batching.