How to switch database connection for the job being processed in laravel

Question

I have a Laravel project that interfaces with a react project via APIs. It uses an RDS MySQL instance as the database.

We perform some pre-computation for the user's social feed via jobs processed through SQS. The database instance usage spikes a lot when these jobs are running and ends up affecting live traffic that connects to the same database instance.

I created a read-replica instance of the original and want my jobs to connect to this read-replica for all computation, primarily all queries that would run should connect to the read-replica connection.

The queries I define within the jobs use the on function to specify which connection to use Model::on('mysql_read_replica') where mysql_read_replica is defined in config/database.php.

However, custom attributes within models, helper functions across the codebase continue to use the original connection. This continues to overload the original instance affecting live traffic.

I have looked at the following question and came up with the following solution: I update config and set connection to the read only replica and change it back at the end of the handle method

public function handle() {
    $mysql_rr = Config::get('database.connections.mysql_rr');
    $mysql = Config::get('database.connections.mysql');
    DB::purge('mysql');
    Config::set('database.connections.mysql', $mysql_rr);

    // job processing code

    DB::purge('mysql');
    Config::set('database.connections.mysql', $mysql);
}

I added an event listening callback and I am able to validate all queries are run against the read replica instance using:

\Event::listen('Illuminate\Database\Events\QueryExecuted', function ($query) {          
    $sql = $query->sql; 
    $time = $query->time;
    $connection = $query->connection->getName();
    $dbName = $query->connection->getDatabaseName();

    Log::info('job query : '.$sql);
    Log::info('job time '.$time);
    Log::info('job connection '.$connection);
    Log::info('job dbName '.$dbName);
});

Since these jobs are supposed to be unique for each user. It has a UniqueJobs trait and the issue I see now is that once the job finishes execution, it fails to release the lock stored on cache. The error log states:

[2022-02-09 13:43:39] local.ERROR: Call to a member function prepare() on null {"exception":"[object] (Error(code: 0): Call to a member function prepare() on null at vendor\laravel\framework\src\Illuminate\Database\Connection.php:492)
[stacktrace]
#0 vendor\laravel\framework\src\Illuminate\Database\Connection.php(671): Illuminate\Database\Connection->Illuminate\Database\{closure}('delete from `ca...', Array)
#1 vendor\laravel\framework\src\Illuminate\Database\Connection.php(638): Illuminate\Database\Connection->runQueryCallback('delete from `ca...', Array, Object(Closure))
#2 vendor\laravel\framework\src\Illuminate\Database\Connection.php(503): Illuminate\Database\Connection->run('delete from `ca...', Array, Object(Closure))
#3 vendor\laravel\framework\src\Illuminate\Database\Connection.php(448): Illuminate\Database\Connection->affectingStatement('delete from `ca...', Array)
#4 vendor\laravel\framework\src\Illuminate\Database\Query\Builder.php(3043): Illuminate\Database\Connection->delete('delete from `ca...', Array)
#5 vendor\laravel\framework\src\Illuminate\Cache\DatabaseLock.php(127): Illuminate\Database\Query\Builder->delete()
#6 vendor\laravel\framework\src\Illuminate\Queue\CallQueuedHandler.php(211): Illuminate\Cache\DatabaseLock->forceRelease()
#7 vendor\laravel\framework\src\Illuminate\Queue\CallQueuedHandler.php(254): Illuminate\Queue\CallQueuedHandler->ensureUniqueJobLockIsReleased(Object(App\Jobs\GenerateDigest))
#8 vendor\laravel\framework\src\Illuminate\Queue\Jobs\Job.php(213): Illuminate\Queue\CallQueuedHandler->failed(Array, Object(Error), '634f580c-3a6f-4...')
#9 vendor\laravel\framework\src\Illuminate\Queue\Jobs\Job.php(192): Illuminate\Queue\Jobs\Job->failed(Object(Error))
#10 vendor\laravel\framework\src\Illuminate\Queue\Worker.php(548): Illuminate\Queue\Jobs\Job->fail(Object(Error))
#11 vendor\laravel\framework\src\Illuminate\Queue\Worker.php(509): Illuminate\Queue\Worker->failJob(Object(Illuminate\Queue\Jobs\SqsJob), Object(Error))
#12 vendor\laravel\framework\src\Illuminate\Queue\Worker.php(437): Illuminate\Queue\Worker->markJobAsFailedIfWillExceedMaxAttempts('sqs', Object(Illuminate\Queue\Jobs\SqsJob), 1, Object(Error))
#13 vendor\laravel\framework\src\Illuminate\Queue\Worker.php(414): Illuminate\Queue\Worker->handleJobException('sqs', Object(Illuminate\Queue\Jobs\SqsJob), Object(Illuminate\Queue\WorkerOptions), Object(Error))
#14 vendor\laravel\framework\src\Illuminate\Queue\Worker.php(360): Illuminate\Queue\Worker->process('sqs', Object(Illuminate\Queue\Jobs\SqsJob), Object(Illuminate\Queue\WorkerOptions))
#15 vendor\laravel\framework\src\Illuminate\Queue\Worker.php(311): Illuminate\Queue\Worker->runJob(Object(Illuminate\Queue\Jobs\SqsJob), 'sqs', Object(Illuminate\Queue\WorkerOptions))
#16 vendor\laravel\framework\src\Illuminate\Queue\Console\WorkCommand.php(117): Illuminate\Queue\Worker->runNextJob('sqs', 'sqs-generation-queue', Object(Illuminate\Queue\WorkerOptions))
#17 vendor\laravel\framework\src\Illuminate\Queue\Console\WorkCommand.php(101): Illuminate\Queue\Console\WorkCommand->runWorker('sqs', 'sqs-generation-queue')
#18 vendor\laravel\framework\src\Illuminate\Container\BoundMethod.php(36): Illuminate\Queue\Console\WorkCommand->handle()
#19 vendor\laravel\framework\src\Illuminate\Container\Util.php(40): Illuminate\Container\BoundMethod::Illuminate\Container\{closure}()
#20 vendor\laravel\framework\src\Illuminate\Container\BoundMethod.php(93): Illuminate\Container\Util::unwrapIfClosure(Object(Closure))
#21 vendor\laravel\framework\src\Illuminate\Container\BoundMethod.php(37): Illuminate\Container\BoundMethod::callBoundMethod(Object(Illuminate\Foundation\Application), Array, Object(Closure))
#22 vendor\laravel\framework\src\Illuminate\Container\Container.php(611): Illuminate\Container\BoundMethod::call(Object(Illuminate\Foundation\Application), Array, Array, NULL)
#23 vendor\laravel\framework\src\Illuminate\Console\Command.php(136): Illuminate\Container\Container->call(Array)
#24 vendor\symfony\console\Command\Command.php(255): Illuminate\Console\Command->execute(Object(Symfony\Component\Console\Input\ArgvInput), Object(Illuminate\Console\OutputStyle))
#25 vendor\laravel\framework\src\Illuminate\Console\Command.php(121): Symfony\Component\Console\Command\Command->run(Object(Symfony\Component\Console\Input\ArgvInput), Object(Illuminate\Console\OutputStyle))
#26 vendor\symfony\console\Application.php(971): Illuminate\Console\Command->run(Object(Symfony\Component\Console\Input\ArgvInput), Object(Symfony\Component\Console\Output\ConsoleOutput))
#27 vendor\symfony\console\Application.php(290): Symfony\Component\Console\Application->doRunCommand(Object(Illuminate\Queue\Console\WorkCommand), Object(Symfony\Component\Console\Input\ArgvInput), Object(Symfony\Component\Console\Output\ConsoleOutput))
#28 vendor\symfony\console\Application.php(166): Symfony\Component\Console\Application->doRun(Object(Symfony\Component\Console\Input\ArgvInput), Object(Symfony\Component\Console\Output\ConsoleOutput))
#29 vendor\laravel\framework\src\Illuminate\Console\Application.php(92): Symfony\Component\Console\Application->run(Object(Symfony\Component\Console\Input\ArgvInput), Object(Symfony\Component\Console\Output\ConsoleOutput))
#30 vendor\laravel\framework\src\Illuminate\Foundation\Console\Kernel.php(129): Illuminate\Console\Application->run(Object(Symfony\Component\Console\Input\ArgvInput), Object(Symfony\Component\Console\Output\ConsoleOutput))
#31 artisan(37): Illuminate\Foundation\Console\Kernel->handle(Object(Symfony\Component\Console\Input\ArgvInput), Object(Symfony\Component\Console\Output\ConsoleOutput))
#32 {main}
"} 

After adding some logging, I find that the getPdo() function in vendor\laravel\framework\src\Illuminate\Database\Connection.php is basically null as the stacktrace says. If I remove the code to switch connections, the job processes fine and is able to release the lock.