40 lines
9.4 KiB
PHP
40 lines
9.4 KiB
PHP
<?php
|
|
use Illuminate\Support\Facades\DB;
|
|
use Illuminate\Support\Str;
|
|
if($phase==='ledger-sales-worker'){
|
|
$input=realpath($argv[3]??'');if(!$input||!str_starts_with($input,realpath($s['root']).DIRECTORY_SEPARATOR))throw new LogicException('Owned worker input required');$w=json_decode(file_get_contents($input),true,64,JSON_THROW_ON_ERROR);
|
|
if(!preg_match('/^sales_(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->sales($enter('read'),$g['invoice_id']);$result=['status'=>200,'balance'=>$data['commercial']['position']['balance'],'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'].'/sales-race.json';if(file_exists($output)||DB::table('sr_records')->exists()||DB::table('gl_journals')->exists())throw new LogicException('Fresh separate race fixture required');
|
|
$r=$sr->propose($enter('prepare')->id,['kind'=>'obligation','invoice_id'=>$g['invoice_id'],'currency'=>'MZN','source'=>'TEST-MANUAL','external_id'=>'SALES-RACE-OPEN','revision'=>0,'reason'=>'Explicit synthetic opening for concurrency.'],(string)Str::uuid());$ob=$r['id'];$sr->decide($ob,$enter('validate')->id,['version'=>1,'revision'=>0,'decision'=>'approved','reason'=>'Independent documentary opening for concurrency.'],(string)Str::uuid());
|
|
$p=$command('prepare','effect-prepare',['effect_type'=>'sales.invoice','source_id'=>$g['invoice_id'],'journal_on'=>'2026-10-29','period_key'=>'QA-2026-10','rule_key'=>'QA-SALES','reason'=>'Capture original three-line invoice for concurrency.']);$command('validate','validate',['proposal_id'=>$p['proposal_id'],'revision'=>$p['revision'],'decision'=>'approved','reason'=>'Independent exact three-line impact review.']);
|
|
$financialChange=function()use($sr,$enter,$g,$ob){$rev=fn()=>DB::table('sr_effects')->where('obligation_id',$ob)->count();$r=$sr->propose($enter('prepare')->id,['kind'=>'receipt','invoice_id'=>$g['invoice_id'],'obligation_id'=>$ob,'currency'=>'MZN','amount'=>'400.00','date'=>'2026-10-29','evidence'=>'Only a synthetic concurrent receipt.','source'=>'TEST-MANUAL','external_id'=>(string)Str::uuid(),'revision'=>$rev(),'reason'=>'Concurrent synthetic financial fact.'],(string)Str::uuid());$sr->decide($r['id'],$enter('validate')->id,['version'=>1,'revision'=>$rev(),'decision'=>'approved','reason'=>'Independent concurrent receipt confirmation.'],(string)Str::uuid());return $sr->allocate($r['id'],$enter('prepare')->id,['version'=>1,'revision'=>$rev(),'amount'=>'400.00'],(string)Str::uuid())['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-sales-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('sales_'.$name.'_'.$n.'_'.$s['tag'],$action,$body);
|
|
if($reader)$readInput=$launch('sales_read_a_'.$s['tag'],'read',[]);
|
|
config(['database.connections.sales_monitor'=>DB::connection()->getConfig()]);$monitor=DB::connection('sales_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=?',['sales_'.$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('sales_monitor');}
|
|
};
|
|
$posting=$race('post','post',['proposal_id'=>$p['proposal_id'],'revision'=>$p['revision'],'reason'=>'Apply original sales journal once under concurrency.'],2,$financialChange,false,true);
|
|
if($posting['read_before_release']['balance']!=='1000.00'||$posting['read_before_release']['journals']!==0||$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()!==3||DB::table('sr_effects')->count()!==1)throw new LogicException('Read isolation or single three-line effect differs');
|
|
$p=$command('prepare','period-prepare',['period_key'=>'QA-2026-10','reason'=>'Capture period before concurrent confirmed receipt.']);$command('validate','period-validate',['check_id'=>$p['check_id'],'revision'=>$p['revision'],'decision'=>'approved','reason'=>'Independent period evidence before new facts.']);
|
|
$stale=$race('stale','period-close',['check_id'=>$p['check_id'],'revision'=>$p['revision'],'reason'=>'Must revalidate confirmed financial changes.'],1,$financialChange,true);
|
|
if($stale['results']['a']['status']!==409||DB::table('gl_period_closures')->exists())throw new LogicException('New confirmed allocation must refuse stale close after release');
|
|
$allocation=$stale['changed_effect_id'];$p=$command('prepare','effect-prepare',['effect_type'=>'receivable.allocated','source_id'=>$allocation,'journal_on'=>'2026-10-29','period_key'=>'QA-2026-10','rule_key'=>'QA-RECEIVABLE','reason'=>'Account newly committed confirmed allocation.']);$command('validate','validate',['proposal_id'=>$p['proposal_id'],'revision'=>$p['revision'],'decision'=>'approved','reason'=>'Independent new allocation accounting review.']);$command('prepare','post',['proposal_id'=>$p['proposal_id'],'revision'=>$p['revision'],'reason'=>'Apply confirmed allocation once.']);
|
|
$p=$command('prepare','period-prepare',['period_key'=>'QA-2026-10','reason'=>'Fresh period capture after financial fact accounted.']);$command('validate','period-validate',['check_id'=>$p['check_id'],'revision'=>$p['revision'],'decision'=>'approved','reason'=>'Independent refreshed commercial period conference.']);$closing=$race('close','period-close',['check_id'=>$p['check_id'],'revision'=>$p['revision'],'reason'=>'Close refreshed artificial period exactly 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 sales close duplicated');
|
|
file_put_contents($output,json_encode(['passed'=>true,'workers'=>6,'three_line_post'=>$posting,'stale_after_confirmed_allocation'=>$stale,'fresh_period_close'=>$closing,'original_invoice_unchanged'=>true,'journals'=>2,'lines'=>5,'financial_balance'=>'600.00','closures'=>1,'read_finished_before_writer_release'=>true],JSON_PRETTY_PRINT|JSON_THROW_ON_ERROR));echo "Six workers: read before release, one three-line journal, stale close409 after committed allocation, fresh close once.\n";
|