37 lines
9.3 KiB
PHP
37 lines
9.3 KiB
PHP
<?php
|
|
use Illuminate\Support\Facades\DB;
|
|
use Illuminate\Support\Str;
|
|
use Tempo\Finance\Payables\Payables;
|
|
if($phase==='ledger-cost-correction-worker'){
|
|
$input=realpath($argv[3]??'');if(!$input||!str_starts_with($input,realpath($s['root']).DIRECTORY_SEPARATOR))throw new LogicException('Owned correction 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'){$d=$ledger->costs($enter('read'),$g['order_id']);$r=['status'=>200,'amount_minor'=>$d['cost']['source']['principal_minor'],'journals'=>count($d['book']['journals']),'results'=>count($d['correction_results'])];}else $r=['status'=>200,'result'=>$command('prepare',$w['action'],$w['body'],$w['key'])];}catch(DomainException $e){$r=['status'=>409,'message'=>$e->getMessage()];}file_put_contents($input.'.result',json_encode($r,JSON_THROW_ON_ERROR));exit;
|
|
}
|
|
$output=$s['root'].'/cost-correction-race.json';if(file_exists($output)||DB::table('gl_corrections')->exists())throw new LogicException('Fresh correction race book required');$prefixOutbox=DB::table('dc_outbox')->orderBy('event_id')->get()->toJson();$secondCalc=hash('sha256',Payables::canonical((array)DB::table('dc_calculations')->where('id',$cg['replacement_calculation_id'])->first()));
|
|
$p=$captureCorrection();$approveCorrection($p);
|
|
$change=function()use($dc,$enter,$g,$cg){$u=$enter('prepare');$r=$dc->rate($g['order_id'],$u->id,['entry_id'=>$g['material_id'],'entry_version'=>1,'identifier'=>'QA-COST-MATERIAL','version'=>3,'previous_id'=>$cg['replacement_rate_id'],'currency'=>'MZN','unit'=>'UN','amount'=>'60.00','evidence'=>'Concurrent version-only rate change, selected amount remains295.'],(string)Str::uuid());$u=$enter('validate');$dc->decide($g['order_id'],$u->id,['kind'=>'rate','id'=>$r['id'],'version'=>3,'decision'=>'approved','reason'=>'Independent concurrent rate version conference.'],(string)Str::uuid());$u=$enter('prepare');$calc=$dc->calculate($g['order_id'],$u->id,['version'=>3,'previous_id'=>$cg['replacement_calculation_id'],'currency'=>'MZN','lines'=>[['entry_id'=>$g['material_id'],'entry_version'=>1,'rate_id'=>$r['id'],'rate_version'=>3],['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'=>3,'decision'=>'approved','reason'=>'Independent current295 with new source version.'],(string)Str::uuid());return $calc['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-correction-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','correct-post',$postCorrectionBody($p),1,$change,true,true);
|
|
if($stale['results']['a']['status']!==409||$stale['read_before_release']['amount_minor']!==29500||$stale['read_before_release']['journals']!==1||$stale['read_before_release']['results']!==0||DB::table('gl_journals')->count()!==1||DB::table('gl_correction_results')->exists())throw new LogicException('Stale correction/read coherence after committed version change failed');
|
|
$p=$captureCorrection();$approveCorrection($p);$applied=$race('post','correct-post',$postCorrectionBody($p));if($applied['results']['a']['status']!==200||$applied['results']['b']['status']!==200||$applied['results']['a']['result']['result_id']!==$applied['results']['b']['result']['result_id']||DB::table('gl_journals')->count()!==3||DB::table('gl_lines')->count()!==6||DB::table('gl_correction_results')->count()!==1)throw new LogicException('Concurrent pair did not recover a single result');
|
|
$p=$command('prepare','period-prepare',['period_key'=>'QA-2026-10','reason'=>'Fresh corrected period after exact version conference.']);$command('validate','period-validate',['check_id'=>$p['check_id'],'revision'=>$p['revision'],'decision'=>'approved','reason'=>'Independent refreshed corrected period.']);$closed=$race('close','period-close',['check_id'=>$p['check_id'],'revision'=>$p['revision'],'reason'=>'Close corrected cost QA period once.']);if($closed['results']['a']['status']!==200||$closed['results']['b']['status']!==200||$closed['results']['a']['result']['closure_id']!==$closed['results']['b']['result']['closure_id']||DB::table('gl_period_closures')->count()!==1)throw new LogicException('Concurrent corrected period duplicated');
|
|
if(hash('sha256',Payables::canonical((array)DB::table('gl_journals')->where('id',$cg['original_journal_id'])->first()))!==$cg['original_journal_sha256']||hash('sha256',Payables::canonical((array)DB::table('dc_calculations')->where('id',$g['calculation_id'])->first()))!==$cg['original_calculation_sha256']||hash('sha256',Payables::canonical((array)DB::table('dc_calculations')->where('id',$cg['replacement_calculation_id'])->first()))!==$secondCalc||DB::table('dc_outbox')->whereIn('event_id',array_column(json_decode($prefixOutbox,true),'event_id'))->orderBy('event_id')->get()->toJson()!==$prefixOutbox||DB::table('dc_outbox')->where('state','<>','captured_no_accounting')->exists())throw new LogicException('Original journal/calculations/outbox changed');
|
|
$report=$ledger->report($enter('read'),'account-ledger',['from_on'=>'2026-10-01','to_on'=>'2026-10-31','account_id'=>$g['accounts']['QA-CUSTO-DIRECTO']]);if($report['closing_net_minor']!==29500)throw new LogicException('Concurrent corrected report does not reconcile');file_put_contents($output,json_encode(['passed'=>true,'workers'=>6,'stale_group_after_committed_version'=>$stale,'fresh_pair'=>$applied,'fresh_close'=>$closed,'reader_finished_before_release'=>true,'prepared_groups'=>2,'applied_groups'=>1,'journals'=>3,'lines'=>6,'closures'=>1,'cost_balance_minor'=>29500,'latest_cost_version'=>3,'original_journal_calculations_outbox_unchanged'=>true],JSON_PRETTY_PRINT|JSON_THROW_ON_ERROR));echo "Six correction workers: read before release, stale409 after committed same-value version, one fresh pair and one close; original journal/calculations/outbox unchanged.\n";
|