33 lines
5.4 KiB
PHP
33 lines
5.4 KiB
PHP
<?php
|
|
if($phase==='ledger-worker'){
|
|
$input=realpath($argv[3]??'');if(!$input||!str_starts_with($input,realpath($s['root']).DIRECTORY_SEPARATOR))throw new LogicException('Private worker input required');$w=json_decode(file_get_contents($input),true,64,JSON_THROW_ON_ERROR);
|
|
if(!preg_match('/^ledger_post_[ab]_[a-f0-9]{12}$/D',$w['label']))throw new LogicException('Worker label refused');
|
|
DB::select('select set_config(?, ?, false)',['application_name',$w['label']]);DB::statement("SET lock_timeout='40s'");DB::statement("SET statement_timeout='45s'");
|
|
try{$result=$command('prepare','post',$w['body'],$w['key']);$out=['status'=>200,'result'=>$result];}
|
|
catch(DomainException $e){$out=['status'=>409,'message'=>$e->getMessage()];}
|
|
file_put_contents($input.'.result',json_encode($out,JSON_PRETTY_PRINT|JSON_THROW_ON_ERROR));exit;
|
|
}
|
|
if(is_file($s['root'].'/ledger-race.json'))throw new LogicException('Completed race preserved');
|
|
if(DB::table('gl_journals')->count()||DB::table('gl_proposals')->count())throw new LogicException('Race requires the retained empty ledger after rollback-only tests');
|
|
$p=$command('prepare','prepare',$g['body']);$command('validate','validate',['proposal_id'=>$p['proposal_id'],'revision'=>$p['revision'],'decision'=>'approved','reason'=>'Independent exact version for concurrency.']);
|
|
$body=['proposal_id'=>$p['proposal_id'],'revision'=>$p['revision'],'reason'=>'Apply the reviewed impact once under concurrency.'];$workers=[];$inputs=[];$keys=[];
|
|
DB::beginTransaction();
|
|
try{
|
|
$parent=(int)DB::selectOne('select pg_backend_pid() AS pid')->pid;
|
|
DB::table('companies')->where('tenant_id',$scope['tenant_id'])->where('id',$scope['company_id'])->lockForUpdate()->firstOrFail();
|
|
foreach(['a','b'] as $name){$input=$s['root'].'/ledger-race-'.$name.'-'.bin2hex(random_bytes(5)).'.json';$keys[$name]=(string)Illuminate\Support\Str::uuid();file_put_contents($input,json_encode(['label'=>'ledger_post_'.$name.'_'.$s['tag'],'key'=>$keys[$name],'body'=>$body]));$log=$input.'.log';$process=proc_open([PHP_BINARY,__DIR__.'/payable-history-sgf-proof.php',$file,'ledger-worker',$input],[0=>['file','NUL','r'],1=>['file',$log,'a'],2=>['file',$log,'a']],$pipes,dirname(__DIR__),null,['bypass_shell'=>true]);if(!is_resource($process))throw new RuntimeException('Owned worker launch failed');$workers[$name]=$process;$inputs[$name]=$input;}
|
|
config(['database.connections.ledger_monitor'=>DB::connection()->getConfig()]);$monitor=DB::connection('ledger_monitor');$observed=[];$blocked=[];$until=microtime(true)+15;
|
|
do{
|
|
$monitor->select('select pg_stat_clear_snapshot()');
|
|
foreach(['a','b'] as $name){$r=$monitor->selectOne('SELECT pid,wait_event_type,wait_event,query,pg_blocking_pids(pid)::text AS blockers FROM pg_stat_activity WHERE datname=current_database() AND application_name=?',['ledger_post_'.$name.'_'.$s['tag']]);if($r)$observed[$name]=['pid'=>(int)$r->pid,'wait'=>$r->wait_event_type,'event'=>$r->wait_event,'blockers'=>json_decode(str_replace(['{','}'],['[',']'],$r->blockers)),'company'=>str_contains($r->query,'"companies"')&&str_contains(strtolower($r->query),'for update')];}
|
|
foreach($observed as $name=>$r){$reaches=in_array($parent,$r['blockers'],true);foreach($observed as $other)if(in_array($other['pid'],$r['blockers'],true)&&in_array($parent,$other['blockers'],true))$reaches=true;if($r['wait']==='Lock'&&$r['company']&&$reaches)$blocked[$name]=['chain_reaches_parent'=>true,'wait_event'=>$r['event'],'statement'=>'companies FOR UPDATE'];}
|
|
if(count($blocked)===2)break;usleep(20000);
|
|
}while(microtime(true)<$until);
|
|
if(count($blocked)!==2)throw new RuntimeException('Two actual posting waits required');
|
|
DB::rollBack();$results=[];
|
|
foreach($inputs as $name=>$input){$deadline=microtime(true)+20;while(!is_file($input.'.result')&µtime(true)<$deadline)usleep(20000);if(!is_file($input.'.result'))throw new RuntimeException('Posting worker did not finish');$results[$name]=json_decode(file_get_contents($input.'.result'),true,64,JSON_THROW_ON_ERROR);}
|
|
if($results['a']['status']!==200||$results['b']['status']!==200||$results['a']['result']['journal_id']!==$results['b']['result']['journal_id']||DB::table('gl_journals')->count()!==1||DB::table('gl_lines')->count()!==2||DB::table('gl_events')->where('kind','journal.posted')->count()!==1)throw new RuntimeException('Concurrent posts duplicated the effect');
|
|
$replay=$command('prepare','post',$body,$keys['a']);if(!$replay['replayed']||$replay['journal_id']!==$results['a']['result']['journal_id']||$financialFingerprint()!==$g['financial_fingerprint'])throw new RuntimeException('Replay or source preservation failed');
|
|
file_put_contents($s['root'].'/ledger-race.json',json_encode(['passed'=>true,'waiting'=>$blocked,'both_completed_after_release'=>true,'same_journal_id'=>true,'journal_id'=>$replay['journal_id'],'journals'=>1,'lines'=>2,'posting_events'=>1,'replay_without_duplicate'=>true,'financial_fingerprint_equal'=>true,'proposal_id'=>$p['proposal_id'],'revision'=>$p['revision'],'post_body'=>$body,'post_key'=>$keys['a']],JSON_PRETTY_PRINT|JSON_THROW_ON_ERROR));echo "Both concurrent postings waited on the company, then recovered one balanced journal; replay adds no effect.\n";
|
|
}finally{while(DB::transactionLevel()>0)DB::rollBack();foreach($workers as $process){$status=proc_get_status($process);if($status['running'])proc_terminate($process);proc_close($process);}DB::purge('ledger_monitor');}
|