[8.x] Delay pushing jobs to queue until database transactions are committed

[themsaid]

DB::transaction(function(){
    $user = User::create(...);

    SendWelcomeEmail::dispatch($user);
});

When running the code above, the SendWelcomeEmail job may get dispatched and picked by a worker before the transaction is committed. This will lead to errors since the user model won't exist when the job runs.

This PR is an attempt to capture queue dispatches, store them in a local cache, and only perform the dispatch when all transactions has been committed. Given the example above, the SendWelcomeEmail won't get dispatched to the queue until the transaction is committed.

To enable this behaviour, you need to set the after_commit configuration value to true in the connection settings inside the queue.php config file. You can also use the afterCommit() method when dispatching the job:

DB::transaction(function(){
    $user = User::create(...);

    SendWelcomeEmail::dispatch($user)->afterCommit();
});

Or if the default is to delay dispatches, you can override that using the beforeCommit() method:

DB::transaction(function(){
    $user = User::create(...);

    SendWelcomeEmail::dispatch($user)->beforeCommit();
});

In the case of rollbacks, the jobs will get discarded.

You can add a dispatchAfterCommit public property to mailables, notifications, listeners, and broadcastable events to achieve the same behaviour.

This PR is an alternative to #35266

[LastDragon-ru]

I'm very glad to see this as built-in functionality, thank you. One question: will the "Queueable" be dispatched in this case?

[themsaid]

@LastDragon-ru no it won't get dispatched. That commit is translated by laravel to a save point.

[LastDragon-ru]

no it won't get dispatched.

Great.

Meanwhile, seems it doesn't work with Queue:fake()

    public function testJobDispatchedAfterTransactionCommitFakeQueue() {
        Queue::fake();

        DB::transaction(function () {
            QueueWithinTransactionJob::dispatch();

            Queue::assertNothingPushed();
            // Jobs were pushed unexpectedly.
            // Failed asserting that an array is empty.
        });

        Queue::assertPushed(QueueWithinTransactionJob::class);
    }

Full TestCase

<?php

namespace Illuminate\Tests\Integration\Queue;

use Exception;
use Illuminate\Bus\Queueable;
use Illuminate\Contracts\Queue\ShouldQueue;
use Illuminate\Foundation\Bus\Dispatchable;
use Illuminate\Support\Facades\DB;
use Illuminate\Support\Facades\Queue;
use Orchestra\Testbench\TestCase;

/**
 * @group integration
 */
class QueueWithinTransactionTest extends TestCase {
    protected function getEnvironmentSetUp($app) {
        $app['config']->set('app.debug', 'true');
        $app['config']->set('database.default', 'mysql');
        $app['config']->set('queue.default', 'redis');
        $app['config']->set('queue.connections.redis.after_commit', true);
    }

    public function testJobDispatchedAfterTransactionCommit() {
        $size = Queue::size();

        DB::transaction(function () use ($size) {
            QueueWithinTransactionJob::dispatch();

            $this->assertEquals($size, Queue::size());
        });

        $this->assertEquals($size + 1, Queue::size());
    }

    public function testJobNotDispatchedAfterRootTransactionRollback() {
        $size = Queue::size();

        try {
            DB::transaction(function () use ($size) {
                DB::transaction(function () use ($size) {
                    QueueWithinTransactionJob::dispatch();

                    $this->assertEquals($size, Queue::size());
                });

                $this->assertEquals($size, Queue::size());

                throw new RollbackException('rollback');
            });
        } catch (RollbackException $exception) {
            // empty
        }

        $this->assertEquals($size, Queue::size());
    }

    public function testJobDispatchedAfterTransactionCommitFakeQueue() {
        Queue::fake();

        DB::transaction(function () {
            QueueWithinTransactionJob::dispatch();

            Queue::assertNothingPushed();
            // Jobs were pushed unexpectedly.
            // Failed asserting that an array is empty.
        });

        Queue::assertPushed(QueueWithinTransactionJob::class);
    }

    public function testJobNotDispatchedAfterRootTransactionRollbackFakeQueue() {
        Queue::fake();

        try {
            DB::transaction(function () {
                QueueWithinTransactionJob::dispatch();

                throw new RollbackException('rollback');
            });
        } catch (RollbackException $exception) {
            // empty
        }

        Queue::assertNothingPushed();
        // Jobs were pushed unexpectedly.
        // Failed asserting that an array is empty.
    }
}

class QueueWithinTransactionJob implements ShouldQueue {
    use Dispatchable, Queueable;

    public function handle() {
    }
}

class RollbackException extends Exception {

}

[themsaid]

Yes it doesn’t have to work when faking the queue, because in your test you are testing the framework not your code.

[LastDragon-ru]

In real app Queue::fake(); also doesn't work :(

<?php declare(strict_types = 1);

namespace Tests\Feature;

use Illuminate\Support\Facades\Queue;
use Tests\TestCase;

class ExampleTest extends TestCase {
    public function testQueueReal() {
        $size = Queue::size();

        $this->get('/test')->assertStatus(503);

        $this->assertEquals($size, Queue::size()); // Passed
    }

    public function testQueueFake() {
        Queue::fake();

        $this->get('/test')->assertStatus(503);

        Queue::assertNothingPushed();
        // Jobs were pushed unexpectedly.
        // Failed asserting that an array is empty.
    }
}

// routes/web.php
Route::get('/test', function () {
    DB::beginTransaction();

    // Doing something ...

    QueueWithinTransactionJob::dispatch();

    // Uops something went wrong...
    DB::rollBack();

    abort(503);
});

// QueueWithinTransactionJob.php
class QueueWithinTransactionJob implements ShouldQueue {
    use Dispatchable, Queueable;

    public $dispatchAfterCommit = true;

    public function handle() {
    }
}

[themsaid]

@LastDragon-ru I'm not sure exactly what your test is supposed to be doing but feel free to open a pull request after/if this one is merged.

Let's keep the discussion focused on the added functionality for now.

[LastDragon-ru]

I'm not sure exactly what your test is supposed to be doing

The test trying to test the added functionality with Sync::fake(), and as I see it is not possible now :(

[fedeisas]

@themsaid will this work for batched jobs? I just ran into a bug using SQS and batches.

DB::transaction(function () use ($import, $iobs) {
    $batch = Bus::batch($jobs)
        ->name('Process Import ' . $import->id)
        ->dispatch();

    $import = $batch->id;
    $import->save();
});

If we want to persiste the batch_id on a related entity within a transaction, SQS jobs start to get processed before the main transaction is committed. Each job atomically moves a couple of columns on job_batches that keep track of the progress.

Each SQS job means an HTTP call (they can be sent in groups of 10 using SendMessageBatch but that's for another PR).

I think the DatabaseBatchRepository::updateAtomicValues won't fail if the batch does not exists. Final scenario is that you get an almost completed job_batch because even if all the jobs ran successfully, some couldn't persist their success/failure anywhere.

[GrahamCampbell]

Needs rebasing on latest 8.x. ;)

[francoism90]

Thanks for adding this, it isn't listed in the docs however? :)

[pabloelcolombiano]

Would the community be interested in a solution that simply gets rid of transactions in the RefreshDatabase trait?

These transactions seem to be creating all kinds of issues. Symfony has the same problem. CakePHP has a solution to refresh the test database that is fast, but without dropping the database at the start of the test suite, thus does not require to run the migrations every time, and cleans up the database based on SQL triggers, without transactions.

I posted this in a discussion #36019 (comment) here, but I am not sure on how to get people of Laravel aware, that cleaning the DB with transactions is not ideal, and that there are other solutions for that.

Here is the repo to the solution for CakePHP https://github.com/vierge-noire/cakephp-test-suite-light
I worked on an adapter for Laravel, which worked like a charm.

I am aware that this PR is not the ideal place to discuss that. I already created a discussion here for that: #36019 (comment)