How to keep db updates in a queued job if timeout occurs?

[victor-david]

Hello All,

Using Laravel 12.15.0, PHP 8.4.14

I have a queued job that sends the same email to a number of users. If the job runs to completion, it may be run again later to send to new users. Pseudo code is like this:

public function handle() 
{
        foreach ($users as $user)
        {
            // checks the database
            if (!$user->hasReceivedThisMail())
            {
               Mail::send(new BulkMailable($this->bulk, $user));
               // updates the database
               $user->updateReceived();
            }   
        }
} 

public function failed(Exception $e)
{
   // can write to the db here without loss because this gets called after the rollback
}

It's a bit more complicated than that, but that's the gist. User gets sent an email and the db is updated to show we sent them that one.

Let's say (for instance), after 100 users, the job times out. Then 100 users get an email, but (because of the timeout) all the updates that indicate who got the email are discarded in a rollback. Therefore, if the job is run again (whether manually or due to Laravel retry settings), those 100 users get a duplicate.

I know I can increase timeout, but the underlying issue is the same IF a timeout occurs.

How can I keep those updates? I've tried wrapping in my own transaction, tried try / catch, but all my tests turn out the same: the updates are discarded.

I think I must be missing something simple, but I don't know what it is. Any help much appreciated.

[macropay-solutions]

Are you dispatching the jobs via Bus::batch()?

[macropay-solutions]

If yes, try to dispatch this job normally without enclosing it into a DB transaction.

[Sandalanka]

This is a classic transactional + long-running job conflict — and your intuition is correct:
Laravel (and the underlying database driver) automatically rolls back all uncommitted database transactions if the job fails, times out, or throws an exception.

If your queued job (or its handle() method) runs inside a database transaction (implicitly or explicitly), and that job fails or times out:

Laravel will rollback all pending changes.

So any $user->updateReceived() calls will not persist, even though the emails were sent.

In Laravel, this often happens because the queue worker wraps jobs in a transaction, depending on configuration.

queue connection configuration (config/queue.php)
'database' => [
'driver' => 'database',
'table' => 'jobs',
'queue' => 'default',
'retry_after' => 90,
'after_commit' => false,
],
Laravel 12, each job can specify whether it should be wrapped in a transaction.

You can disable transactional behavior for this specific job by adding:

class SendBulkMailJob implements ShouldQueue
{
use Dispatchable, InteractsWithQueue, Queueable, SerializesModels;

public $afterCommit = false; // <– critical

public bool $tries = 3;

public function handle()
{
    foreach ($this->users as $user) {
        if (!$user->hasReceivedThisMail()) {
            Mail::send(new BulkMailable($this->bulk, $user));
            $user->updateReceived();
        }
    }
}

public function failed(Exception $e)
{
    // still called after rollback — but your updates won’t rollback now
}

}
Or Make DB Updates “Non-Transactional”

// Default connection (maybe transactional)
DB::transaction(function () use ($users) {
foreach ($users as $user) {
if (!$user->hasReceivedThisMail()) {
Mail::send(new BulkMailable($this->bulk, $user));
// Use a non-transactional connection
DB::connection('mysql_non_transactional')
->table('user_mail_log')
->updateOrInsert(
['user_id' => $user->id, 'mail_id' => $this->bulk->id],
['sent_at' => now()]
);
}
}
});

[macropay-solutions]

Finally \Illuminate\Queue\Jobs\Job::fail

see the rollback:

    /**
     * Delete the job, call the "failed" method, and raise the failed job event.
     *
     * @param  \Throwable|null  $e
     * @return void
     */
    public function fail($e = null)
    {
        $this->markAsFailed();

        if ($this->isDeleted()) {
            return;
        }

        $commandName = $this->payload()['data']['commandName'] ?? false;

        // If the exception is due to a job timing out, we need to rollback the current
        // database transaction so that the failed job count can be incremented with
        // the proper value. Otherwise, the current transaction will never commit.
        if ($e instanceof TimeoutExceededException &&
            $commandName &&
            in_array(Batchable::class, class_uses_recursive($commandName))) {
            $batchRepository = $this->resolve(BatchRepository::class);

            try {
                $batchRepository->rollBack();
            } catch (Throwable $e) {
                // ...
            }
        }

        if ($this->shouldRollBackDatabaseTransaction($e)) {
            $this->container->make('db')
                ->connection($this->container['config']['queue.failed.database'])
                ->rollBack(toLevel: 0);
        }

        try {
            // If the job has failed, we will delete it, call the "failed" method and then call
            // an event indicating the job has failed so it can be logged if needed. This is
            // to allow every developer to better keep monitor of their failed queue jobs.
            $this->delete();

            $this->failed($e);
        } finally {
            $this->resolve(Dispatcher::class)->dispatch(new JobFailed(
                $this->connectionName, $this, $e ?: new ManuallyFailedException
            ));
        }
    }

[macropay-solutions]

This code is extra vs laravel 10

        if ($this->shouldRollBackDatabaseTransaction($e)) {
            $this->container->make('db')
                ->connection($this->container['config']['queue.failed.database'])
                ->rollBack(toLevel: 0);
        }

    protected function shouldRollBackDatabaseTransaction($e)
    {
        return $e instanceof TimeoutExceededException &&
            $this->container['config']['queue.failed.database'] &&
            in_array($this->container['config']['queue.failed.driver'], ['database', 'database-uuids']) &&
            $this->container->bound('db');
    }

[macropay-solutions]

commit 08c2101
pull #54173
issue #49389

This comment mentions that batchable uses transaction #49389 (comment).

[macropay-solutions]

@victor-david you found 2 issues. One is linked to the above fixed issue and one if the fact that the Redis cache key is not removed.

This is the code responsible for releasing the lock
\Illuminate\Queue\Middleware\WithoutOverlapping::handle

    public function handle($job, $next)
    {
        $lock = Container::getInstance()->make(Cache::class)->lock(
            $this->getLockKey($job), $this->expiresAfter
        );

        if ($lock->get()) {
            try {
                $next($job);
            } finally {
                $lock->release();
            }
        } elseif (! is_null($this->releaseAfter)) {
            $job->release($this->releaseAfter);
        }
    }

This is why the finally does not execute

    protected function registerTimeoutHandler($job, WorkerOptions $options)
    {
        // We will register a signal handler for the alarm signal so that we can kill this
        // process if it is running too long because it has frozen. This uses the async
        // signals supported in recent versions of PHP to accomplish it conveniently.
        pcntl_signal(SIGALRM, function () use ($job, $options) {
            if ($job) {
                $this->markJobAsFailedIfWillExceedMaxAttempts(
                    $job->getConnectionName(),
                    $job,
                    (int)$options->maxTries,
                    $e = $this->timeoutExceededException($job)
                );

                $this->markJobAsFailedIfWillExceedMaxExceptions(
                    $job->getConnectionName(),
                    $job,
                    $e
                );

                $this->markJobAsFailedIfItShouldFailOnTimeout(
                    $job->getConnectionName(),
                    $job,
                    $e
                );

                $this->events->dispatch(
                    new JobTimedOut(
                        $job->getConnectionName(),
                        $job
                    )
                );
            }

            $this->kill(static::EXIT_ERROR, $options);
        }, true);

        pcntl_alarm(
            max($this->timeoutForJob($job, $options), 0)
        );
    }

    public function kill($status = 0, $options = null)
    {
        $this->events->dispatch(new WorkerStopping($status, $options));

        if (extension_loaded('posix')) {
            posix_kill(getmypid(), SIGKILL);
        }

        exit($status);
    }

[macropay-solutions]

But the rollback is linked to your job using Batchable trait.

Please confirm @victor-david

[macropay-solutions]

@victor-david use a child function of WithoutOverlaping as middleware and in it override the handle method:

    public function handle($job, $next)
    {
        $lock = Container::getInstance()->make(Cache::class)->lock(
            $this->getLockKey($job),
            $this->expiresAfter
        );

        if ($lock->get()) {
            $lockWrapper = new \stdClass();
            $lockWrapper->lock = $lock;

            try {
                \register_shutdown_function(function () use ($lockWrapper): void {
                    if (\isset($lockWrapper->lock)) {
                        $lockWrapper->lock->release();
                    }
                });

                $next($job);
            } finally {
                $lock->release();
                $lockWrapper->lock = null;
            }
        } elseif (!is_null($this->releaseAfter)) {
            $job->release($this->releaseAfter);
        }
    }

Run your queue worker with limited amount of jobs to be processed in a PID to avoid the shutdown functions piling up and eating memory.

php artisan queue:work --max-jobs=50

[macropay-solutions]

This is affecting also the ShouldBeUnique and not ShouldBeUniqueUntilProcessing when the worker exists due to timeout... Basically any job that sets a lock in cache is affected by this.
Because the default expiration for lock is 0 that means the lock will never expire.

[macropay-solutions]

Best solution is to ALWAYS set expiration times on the lock because this can happen also on a deploy or on something unexpected.

public function middleware(): array
{
    return [new WithoutOverlapping($this->user->id, 0, 3600)];
}

For ShouldBeunique jobs property or method are required

        $uniqueFor = method_exists($job, 'uniqueFor')
            ? $job->uniqueFor()
            : ($job->uniqueFor ?? 0);

See commit macropay-solutions/maravel-framework@29a2d75

And thank you.

[victor-david]

I was thinking about extending WithoutOverlapping and do something different with the lock (although I wasn't sure what). Thanks for the tips and directions. I'll try these things out.