41 lines
9.4 KiB
PHP
41 lines
9.4 KiB
PHP
<?php
|
|
use Illuminate\Support\Facades\DB;
|
|
use Illuminate\Support\Str;
|
|
use Tempo\Finance\Payables\Payables;
|
|
if($phase==='ledger-cost-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('/^cost_(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'){$data=$ledger->costs($enter('read'),$g['order_id']);$result=['status'=>200,'amount_minor'=>$data['cost']['source']['principal_minor'],'journals'=>count($data['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'].'/cost-race.json';if(file_exists($output)||DB::table('gl_proposals')->exists())throw new LogicException('Fresh separate cost race fixture required');
|
|
$originalCalc=hash('sha256',Payables::canonical((array)DB::table('dc_calculations')->where('id',$g['calculation_id'])->first()));$technical=function(){$h=hash_init('sha256');foreach(['wo_orders','wo_entries','wo_decisions'] as $t)foreach(DB::table($t)->orderBy('id')->cursor() as $r)hash_update($h,Payables::canonical((array)$r));return hash_final($h);};$technicalBefore=$technical();$oldOutbox=DB::table('dc_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'=>'Independent original cost review before race.']);
|
|
$change=function()use($dc,$enter,$g){$u=$enter('prepare');$r=$dc->rate($g['order_id'],$u->id,['entry_id'=>$g['material_id'],'entry_version'=>1,'identifier'=>'QA-COST-MATERIAL','version'=>2,'previous_id'=>$g['rate_ids']['material'],'currency'=>'MZN','unit'=>'UN','amount'=>'50.00','evidence'=>'Concurrent successor has same value but another actual reviewed version.'],(string)Str::uuid());$u=$enter('validate');$dc->decide($g['order_id'],$u->id,['kind'=>'rate','id'=>$r['id'],'version'=>2,'decision'=>'approved','reason'=>'Independent concurrent successor rate approval.'],(string)Str::uuid());return $r['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-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('cost_'.$name.'_'.$n.'_'.$s['tag'],$action,$body);
|
|
if($reader)$readInput=$launch('cost_read_a_'.$s['tag'],'read',[]);
|
|
config(['database.connections.cost_monitor'=>DB::connection()->getConfig()]);$monitor=DB::connection('cost_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=?',['cost_'.$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('cost_monitor');}
|
|
};
|
|
|
|
$stale=$race('stale','post',['proposal_id'=>$p['proposal_id'],'revision'=>$p['revision'],'reason'=>'Must revalidate committed cost source version.'],1,$change,true,true);
|
|
if($stale['read_before_release']['amount_minor']!==27500||$stale['read_before_release']['journals']!==0||$stale['results']['a']['status']!==409||DB::table('gl_journals')->exists())throw new LogicException('Reader isolation or stale cost version refusal failed');
|
|
$newRate=$stale['changed_effect_id'];$u=$enter('prepare');$calc=$dc->calculate($g['order_id'],$u->id,['version'=>2,'previous_id'=>$g['calculation_id'],'currency'=>'MZN','lines'=>[['entry_id'=>$g['material_id'],'entry_version'=>1,'rate_id'=>$newRate,'rate_version'=>2],['entry_id'=>$g['time_id'],'entry_version'=>1,'rate_id'=>$g['rate_ids']['time'],'rate_version'=>1]]],(string)Str::uuid());$u=$enter('validate');$dc->decide($g['order_id'],$u->id,['kind'=>'calculation','id'=>$calc['id'],'version'=>2,'decision'=>'approved','reason'=>'Independent refreshed calculation275 after exact source version changed.'],(string)Str::uuid());
|
|
$p=$command('prepare','effect-prepare',$body);$command('validate','validate',['proposal_id'=>$p['proposal_id'],'revision'=>$p['revision'],'decision'=>'approved','reason'=>'Independent refreshed accounting capture.']);$posting=$race('post','post',['proposal_id'=>$p['proposal_id'],'revision'=>$p['revision'],'reason'=>'Apply refreshed selected cost 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')->count()!==1||DB::table('gl_lines')->count()!==2)throw new LogicException('Concurrent cost posting duplicated or did not recover');
|
|
$p=$command('prepare','period-prepare',['period_key'=>'QA-2026-10','reason'=>'Confer current selected cost after source refresh and one posting.']);$command('validate','period-validate',['check_id'=>$p['check_id'],'revision'=>$p['revision'],'decision'=>'approved','reason'=>'Independent period conference.']);$closing=$race('close','period-close',['check_id'=>$p['check_id'],'revision'=>$p['revision'],'reason'=>'Close refreshed selected cost QA once.']);
|
|
if($closing['results']['a']['status']!==200||$closing['results']['b']['status']!==200||$closing['results']['a']['result']['closure_id']!==$closing['results']['b']['result']['closure_id']||DB::table('gl_period_closures')->count()!==1||DB::table('gl_periods')->count()!==2)throw new LogicException('Concurrent period close duplicated');
|
|
if($technical()!==$technicalBefore||hash('sha256',Payables::canonical((array)DB::table('dc_calculations')->where('id',$g['calculation_id'])->first()))!==$originalCalc||DB::table('dc_outbox')->whereIn('event_id',array_column(json_decode($oldOutbox,true),'event_id'))->orderBy('event_id')->get()->toJson()!==$oldOutbox||DB::table('dc_outbox')->where('state','<>','captured_no_accounting')->exists())throw new LogicException('Original technical/cost/outbox state changed');
|
|
file_put_contents($output,json_encode(['passed'=>true,'workers'=>6,'stale_post_after_committed_rate_version'=>$stale,'fresh_posting'=>$posting,'fresh_period_close'=>$closing,'read_finished_before_writer_release'=>true,'same_cost_amount_minor'=>27500,'new_calculation_version'=>2,'journals'=>1,'lines'=>2,'closures'=>1,'technical_original_calculation_and_original_outbox_unchanged'=>true,'all_outbox_still_captured_no_accounting'=>true],JSON_PRETTY_PRINT|JSON_THROW_ON_ERROR));echo "Six cost workers: reader before release, stale409 after committed source version, refreshed post and close each once; originals/outbox preserved.\n";
|