36 lines
8.7 KiB
PHP
36 lines
8.7 KiB
PHP
<?php
|
|
use Illuminate\Support\Facades\DB;
|
|
use Illuminate\Support\Str;
|
|
if($phase==='ledger-cost-industrial-worker'){
|
|
$input=realpath($argv[3]??'');if(!$input||!str_starts_with($input,realpath($s['root']).DIRECTORY_SEPARATOR))throw new LogicException('Owned worker required');$w=json_decode(file_get_contents($input),true,64,JSON_THROW_ON_ERROR);if(!preg_match('/^industrial_(post|close|stale|read)_[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='30s'");try{if($w['action']==='read'){$d=$ledger->industrial($enter('read'),$i['lot_id']);$result=['status'=>200,'amount_minor'=>$d['industrial']['source']['principal_minor'],'journals'=>count($d['book']['journals'])];}else $result=['status'=>200,'result'=>$command('prepare',$w['action'],$w['body'],$w['key'])];}catch(DomainException $e){$result=['status'=>409,'message'=>$e->getMessage()];}file_put_contents($input.'.result',json_encode($result,JSON_THROW_ON_ERROR));exit;
|
|
}
|
|
$output=$s['root'].'/industrial-race.json';if(file_exists($output)||DB::table('gl_journals')->where('snapshot->source->effect_type','industry.cost.validated')->exists())throw new LogicException('Fresh industrial race book');
|
|
$originalResult=DB::table('ic_results')->where('id',$i['result_id'])->value('fingerprint');$outbox=DB::table('ic_outbox')->orderBy('event_id')->get()->toJson();
|
|
$p=$command('prepare','effect-prepare',$body);$command('validate','validate',['proposal_id'=>$p['proposal_id'],'revision'=>$p['revision'],'decision'=>'approved','reason'=>'Distinct original industrial impact conference.']);
|
|
$change=function()use($ic,$i,$enter){$r=DB::table('ic_inputs')->find($i['refs']['material_rate']['id']);$data=json_decode($r->data,true);$r2=$ic->input($i['lot_id'],$enter('prepare')->id,['kind'=>'material_rate','source_ref'=>$r->source_ref,'version'=>2,'previous_id'=>$r->id,'data'=>$data,'evidence'=>'Concurrent exact successor rate50, same value.'],(string)Str::uuid());$ic->decide($i['lot_id'],$enter('validate')->id,['type'=>'input','id'=>$r2['id'],'version'=>2,'decision'=>'approved','reason'=>'Distinct concurrent industrial successor rate.'],(string)Str::uuid());return $r2['id'];};
|
|
$race=function($name,$action,$body,$number=2,$change=null,$commit=false,$reader=false)use($s,$file,$scope){
|
|
$workers=[];$inputs=[];$readInput=null;$observed=[];DB::beginTransaction();try{
|
|
$parent=(int)DB::selectOne('SELECT pg_backend_pid() pid')->pid;DB::table('companies')->where('id',$scope['company_id'])->where('tenant_id',$scope['tenant_id'])->lockForUpdate()->firstOrFail();$changed=$change?$change():null;
|
|
$launch=function($label,$workerAction,$body)use(&$workers,$s,$file){$input=$s['root'].'/'.$label.'-'.bin2hex(random_bytes(4)).'.json';file_put_contents($input,json_encode(['label'=>$label,'action'=>$workerAction,'body'=>$body,'key'=>(string)Str::uuid()]));$log=$input.'.log';$p=proc_open([PHP_BINARY,__DIR__.'/payable-history-sgf-proof.php',$file,'ledger-cost-industrial-worker',$input],[0=>['file','NUL','r'],1=>['file',$log,'a'],2=>['file',$log,'a']],$pipes,dirname(__DIR__),null,['bypass_shell'=>true]);if(!is_resource($p))throw new LogicException('Worker launch failed');$workers[]=$p;return $input;};
|
|
foreach(array_slice(['a','b'],0,$number) as $n)$inputs[$n]=$launch('industrial_'.$name.'_'.$n.'_'.$s['tag'],$action,$body);
|
|
if($reader)$readInput=$launch('industrial_read_a_'.$s['tag'],'read',[]);
|
|
config(['database.connections.industrial_monitor'=>DB::connection()->getConfig()]);$monitor=DB::connection('industrial_monitor');$blocked=[];$until=microtime(true)+15;
|
|
do{$monitor->select('SELECT pg_stat_clear_snapshot()');foreach($inputs as $n=>$input){$r=$monitor->selectOne('SELECT pid,wait_event_type,wait_event,query,pg_blocking_pids(pid)::text blockers FROM pg_stat_activity WHERE datname=current_database() AND application_name=?',['industrial_'.$name.'_'.$n.'_'.$s['tag']]);if($r)$observed[$n]=['pid'=>(int)$r->pid,'wait'=>$r->wait_event_type,'event'=>$r->wait_event,'company'=>str_contains($r->query,'"companies"')&&str_contains(strtolower($r->query),'for update'),'blockers'=>json_decode(str_replace(['{','}'],['[',']'],$r->blockers))];}
|
|
foreach($observed as $n=>$r){$reach=in_array($parent,$r['blockers'],true);foreach($observed as $other)if(in_array($other['pid'],$r['blockers'],true)&&in_array($parent,$other['blockers'],true))$reach=true;if($reach&&$r['wait']==='Lock'&&$r['company'])$blocked[$n]=['chain_reaches_parent'=>true,'wait_event'=>$r['event']];}
|
|
if(count($blocked)===$number&&(!$readInput||file_exists($readInput.'.result')))break;usleep(20000);
|
|
}while(microtime(true)<$until);
|
|
if(count($blocked)!==$number||($readInput&&!file_exists($readInput.'.result')))throw new LogicException('Actual blocked writers and independent reader required');
|
|
$readResult=$readInput?json_decode(file_get_contents($readInput.'.result'),true,64,JSON_THROW_ON_ERROR):null;
|
|
$commit?DB::commit():DB::rollBack();$results=[];foreach($inputs as $n=>$input){$until=microtime(true)+20;while(!file_exists($input.'.result')&µtime(true)<$until)usleep(20000);if(!file_exists($input.'.result'))throw new LogicException('Worker did not finish after release');$results[$n]=json_decode(file_get_contents($input.'.result'),true,64,JSON_THROW_ON_ERROR);}
|
|
return ['waiting'=>$blocked,'read_before_release'=>$readResult,'results'=>$results,'changed_effect_id'=>$commit?$changed:null];
|
|
}finally{while(DB::transactionLevel()>0)DB::rollBack();foreach($workers as $p){if(proc_get_status($p)['running'])proc_terminate($p);proc_close($p);}DB::purge('industrial_monitor');}
|
|
};
|
|
|
|
$stale=$race('stale','post',['proposal_id'=>$p['proposal_id'],'revision'=>$p['revision'],'reason'=>'Must refuse changed industrial version after release.'],1,$change,true,true);
|
|
if($stale['results']['a']['status']!==409||$stale['read_before_release']['amount_minor']!==180000||$stale['read_before_release']['journals']!==0)throw new LogicException('Industrial stale refusal or nonblocking snapshot reader failed');
|
|
$refs=$i['refs'];$refs['material_rate']=['id'=>$stale['changed_effect_id'],'version'=>2];$r=$ic->calculate($i['lot_id'],$enter('prepare')->id,['version'=>2,'previous_id'=>$i['result_id'],'refs'=>$refs,'currency'=>'MZN'],(string)Str::uuid());$ic->decide($i['lot_id'],$enter('validate')->id,['type'=>'result','id'=>$r['id'],'version'=>2,'decision'=>'approved','reason'=>'Current result1800 with refreshed exact versions.'],(string)Str::uuid());$p=$command('prepare','effect-prepare',$body);$command('validate','validate',['proposal_id'=>$p['proposal_id'],'revision'=>$p['revision'],'decision'=>'approved','reason'=>'Distinct refreshed industrial accounting impact.']);$posting=$race('post','post',['proposal_id'=>$p['proposal_id'],'revision'=>$p['revision'],'reason'=>'Apply refreshed industrial1800 once.']);
|
|
if($posting['results']['a']['status']!==200||$posting['results']['b']['status']!==200||$posting['results']['a']['result']['journal_id']!==$posting['results']['b']['result']['journal_id']||DB::table('gl_journals')->where('snapshot->source->effect_type','industry.cost.validated')->count()!==1)throw new LogicException('Industrial concurrent post did not recover one effect');
|
|
$closing=null;$compositionCloseRace=function($body)use($race,&$closing){$closing=$race('close','period-close',$body);if($closing['results']['a']['status']!==200||$closing['results']['b']['status']!==200||$closing['results']['a']['result']['closure_id']!==$closing['results']['b']['result']['closure_id'])throw new LogicException('Joint industrial close duplicated or refused');};
|
|
$cf=json_decode(file_get_contents($s['root'].'/ledger-composition-fixture.json'),true);require __DIR__.'/ledger-composition-workflow.php';
|
|
if(DB::table('gl_journals')->count()!==9||DB::table('gl_lines')->count()!==19||DB::table('gl_period_closures')->count()!==1||DB::table('ic_results')->where('id',$i['result_id'])->value('fingerprint')!==$originalResult||DB::table('ic_outbox')->whereIn('event_id',array_column(json_decode($outbox,true),'event_id'))->orderBy('event_id')->get()->toJson()!==$outbox||DB::table('ic_outbox')->where('state','<>','captured_no_stock_or_accounting')->exists())throw new LogicException('Final industrial race/source/outbox totals differ');
|
|
file_put_contents($output,json_encode(['passed'=>true,'workers'=>6,'stale'=>$stale,'posting'=>$posting,'joint_closing'=>$closing,'journals'=>9,'lines'=>19,'gross_minor'=>510500,'closures'=>1,'original_result_and_outbox_preserved'=>true,'residual_exact'=>'300','reader_finished_before_release'=>true],JSON_PRETTY_PRINT|JSON_THROW_ON_ERROR));echo "Six industrial workers: independent reader, stale409 after source commit, refreshed single effect and joint single close5105.\n"; |