Laravel - Notify or dispatch job queue worker laravel can cause double select with lock table?

user_14180006 0
Viewed 771

Can two Laravel workers can use the same Transaction DB?

I have Job Process A which will call/dispatch Job Process B if there is data in table A with flag is_processed = 0. What it does is:

-- first select data with lock
SELECT *
FROM tableA
WHERE is_proccesed = 0
LIMIT 1000
FOR UPDATE OF tableA SKIP LOCKED

-- insert data to tableB 
INSERT tableB VALUES SELECT values from tableA
    
-- update data 
UPDATE tableA SET is_proccesed = 1 where id = (from any id i have select)

Then trigger job process B:

ProcessB::dispatch(from any id i have select as string)->onQueue('queueA');

I have Job Process B which will be triggered by Job Process A or cron which works every minute.

--first select data with lock
   
SELECT *
FROM tableB
WHERE is_proccesed = 0 AND id in (parameter get from job A if any)
LIMIT 1000
FOR UPDATE OF tableB SKIP LOCKED
    
-- call API with parameter value is from tableB
-- update data

If (call API is success) then:

UPDATE tableB SET is_proccesed = 1 where id = (from any id i have select)

if (call API is fail) then:

UPDATE tableB SET is_proccesed = 0 where id = (from any id i have select)
  • I have a cron running every minute that will call/dispatch Job Process A if any is_processed flag is 0 in table A.
  • I have a cron running every minute which call/dispatch Job Process B if there is_processed flag which is 0 in table B.
  • I use supervisor to do this in real time and use max-retry for jobs that fail 3 times.

My problem is:

  • I have double process call API from job process B,
  • I have scrolled through my logs and the SELECT key got 2 data from 2 different processes at the same time. (in some cases with 2000 or more data to process),
  • It doesn't always happen to process a bit of data.

My question is:

  • Is select data with lock not working with queue jobs?
  • Is it correct to create cron to notify job manually to reprocess unsuccessful data, or should I apply a failed job only to rework jobs?
laravel queue cron postgresql php
1 Answers

I have not seen many Web languages that use database locks correctly. Without looking at the Laravel code, I would guess that it does not use database locks correctly for jobs. I know that it does not use locks for migrate. Running migrate from >2 web nodes is not safe.

If you use Redis or some other technology for jobs instead of SQL DB, a lot of concurrent problems will probably go away.

Manage your own global lock

You can manage your own lock and add synchronization between your own processes.

    $results = \DB::select('SELECT GET_LOCK("process-b", 120) as obtain_lock');
    if (!$results[0]->obtain_lock) { return 0; }
    //120 is seconds to wait for lock or fail

    //load one record

    //call API

    //update one record

    //free lock
    $results = \DB::select('SELECT RELEASE_LOCK("process-b")');
    if (!$results[0]->obtain_lock) { return -1; } //couldn't release lock, stop process, free mysql connection

In Postgresql they are called "advisory locks", but you cannot use characters, you have to use numbers

    $results = \DB::select('SELECT pg_advisory_lock(1337)');
    if (!$results) { return 0; }  // ???

    //load one record

    //call API

    //update one record

    //free lock
    $results = \DB::select('SELECT pg_advisory_unlock(1337)');
    if (!$results) { return -1; } //??? how to check for success?

Use "SELECT ... FOR UPDATE"

I'm not sure if you are trying to use FOR UPDATE locks and it is not working, or you are skipping the lock with intention.

You need to turn off autocommit (set autocommit=0) to use lock FOR UPDATE or to start a transaction.

\DB::transaction( function () use ($id) {
    $results = \DB::table('table_b')->select('SELECT * from table_b where ID=?', $id)->lockForUpdate()->get();
    \DB::table('table_b')->update('UPDATE table_b set x=y where ID=?', $id);
});

Where ProcA sends jobs to ProcB, you can make 1 ProcB job for each ID that is processed=0 - OR - you can make 1 ProcB job whenever you find any processed=0 records.

So, if ProcB will only work with 1 record ID, then global lock solution is probably not good.

You can check that your lock for update is working by putting sleep() and creating 10-20 ProcB jobs with the same record ID. If you sleep for 3 seconds, and it takes 30-60 seconds to finish all ProcB jobs, then the lock for update is working properly. If they all finish in 3 seconds, then they are not respecting the lock on the record.

Bonus

Add this to your routes/console.php to get concurrent-safe artisan lockingmigrate command

$signature = 'lockingmigrate {--database= : The database connection to use}
                {--force : Force the operation to run when in production}
                {--path=* : The path(s) to the migrations files to be executed}
                {--realpath : Indicate any provided migration file paths are pre-resolved absolute paths}
                {--pretend : Dump the SQL queries that would be run}
                {--seed : Indicates if the seed task should be re-run}
                {--step : Force the migrations to be run so they can be rolled back individually}';

Artisan::command($signature, function ($database=false, $seed=false, $step=false, $pretend=false, $force=false) {
    $results = \DB::select('SELECT GET_LOCK("artisan-migrate", 120) as migrate');
    if (!$results[0]->migrate) { return -1; }
    $params = [
        '--pretend' => $pretend,
        '--force'   => $force,
        '--step'    => $step,
        '--seed'    => $seed,
    ];
    $retval = Artisan::call('migrate', $params);
    $outputLines = explode("\n", trim(\Artisan::output()));
    dump($outputLines);
    \DB::select('SELECT RELEASE_LOCK("artisan-migrate")');
    return $retval;
})->describe('Concurrent-safe migrate');
user_3806660
Related