37 lines
6.5 KiB
PHP
37 lines
6.5 KiB
PHP
<?php
|
|
use Illuminate\Support\Str;
|
|
if($phase==='ledger-cycle-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('/^cycle_(correct|close|stale)_[ab]_[a-f0-9]{12}$/D',$w['label'])||!in_array($w['action'],['correct-post','period-close'],true))throw new LogicException('Worker request refused');
|
|
DB::select('SELECT set_config(?,?,false)',['application_name',$w['label']]);DB::statement("SET lock_timeout='30s'");
|
|
try{$r=$command('prepare',$w['action'],$w['body'],$w['key']);$result=['status'=>200,'result'=>$r];}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'].'/ledger-cycle-race.json';if(file_exists($output))throw new LogicException('Completed race preserved');$prepared=json_decode(file_get_contents($s['root'].'/ledger-cycle-race-prepared.json'),true,64,JSON_THROW_ON_ERROR);
|
|
$race=function(string $name,string $action,array $body,int $number=2,?Closure $change=null)use($s,$file){
|
|
$workers=[];$inputs=[];$observed=[];DB::beginTransaction();
|
|
try{
|
|
$parent=(int)DB::selectOne('SELECT pg_backend_pid() pid')->pid;DB::table('companies')->where('id',1)->lockForUpdate()->firstOrFail();if($change)$change();
|
|
foreach(array_slice(['a','b'],0,$number) as $n){$input=$s['root'].'/cycle-'.$name.'-'.$n.'-'.bin2hex(random_bytes(4)).'.json';file_put_contents($input,json_encode(['label'=>'cycle_'.$name.'_'.$n.'_'.$s['tag'],'action'=>$action,'body'=>$body,'key'=>(string)Str::uuid()]));$log=$input.'.log';$p=proc_open([PHP_BINARY,__DIR__.'/payable-history-sgf-proof.php',$file,'ledger-cycle-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[$n]=$p;$inputs[$n]=$input;}
|
|
config(['database.connections.cycle_monitor'=>DB::connection()->getConfig()]);$monitor=DB::connection('cycle_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=?',['cycle_'.$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($r['wait']==='Lock'&&$r['company']&&$reach)$blocked[$n]=['chain_reaches_parent'=>true,'event'=>$r['event']];}
|
|
if(count($blocked)===$number)break;usleep(20000);
|
|
}while(microtime(true)<$until);
|
|
if(count($blocked)!==$number)throw new LogicException('Actual company lock waits required');
|
|
$change?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 complete after release');$results[$n]=json_decode(file_get_contents($input.'.result'),true,64,JSON_THROW_ON_ERROR);}
|
|
return ['blocked'=>$blocked,'results'=>$results];
|
|
}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('cycle_monitor');}
|
|
};
|
|
$correct=$race('correct','correct-post',$prepared['correction_body']);
|
|
if($correct['results']['a']['status']!==200||$correct['results']['b']['status']!==200||$correct['results']['a']['result']['result_id']!==$correct['results']['b']['result']['result_id']||DB::table('gl_correction_results')->count()!==1||DB::table('gl_journals')->count()!==5)throw new LogicException('Concurrent corrective pair duplicated');
|
|
$p=$command('prepare','period-prepare',['period_key'=>'QA-2026-10','reason'=>'Capture before concurrent evidence change.']);$command('validate','period-validate',['check_id'=>$p['check_id'],'revision'=>$p['revision'],'decision'=>'approved','reason'=>'Independent pre-change close review.']);
|
|
$stale=$race('stale','period-close',['check_id'=>$p['check_id'],'revision'=>$p['revision'],'reason'=>'Must revalidate after writer commits changed rule.'],1,function(){ $r=(array)DB::table('gl_rules')->where('rule_key','QA-ALLOCATION')->first();unset($r['id']);$r['version']++;$r['reason']='Explicit synthetic concurrent rule version.';DB::table('gl_rules')->insert($r); });
|
|
if($stale['results']['a']['status']!==409||DB::table('gl_period_closures')->exists())throw new LogicException('Changed evidence must refuse closing after lock release');
|
|
$p=$command('prepare','period-prepare',['period_key'=>'QA-2026-10','reason'=>'Fresh conference after actual concurrent change.']);$command('validate','period-validate',['check_id'=>$p['check_id'],'revision'=>$p['revision'],'decision'=>'approved','reason'=>'Independent review of updated period evidence.']);
|
|
$closeBody=['check_id'=>$p['check_id'],'revision'=>$p['revision'],'reason'=>'Apply updated artificial period close once.'];$closed=$race('close','period-close',$closeBody);
|
|
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||DB::table('gl_periods')->count()!==2)throw new LogicException('Concurrent close duplicated version');
|
|
DB::select('SELECT gl_assert_balanced(id) FROM gl_journals');
|
|
file_put_contents($output,json_encode(['passed'=>true,'corrective_pair'=>$correct,'stale_after_committed_change'=>$stale,'period_close'=>$closed,'journals'=>5,'lines'=>10,'correction_results'=>1,'closures'=>1,'period_versions'=>2,'old_check_id'=>$stale['results']['a']['status'],'final_check_id'=>$p['check_id'],'final_revision'=>$p['revision']],JSON_PRETTY_PRINT|JSON_THROW_ON_ERROR));echo "Five real workers: corrective pair once, stale close409 after committed change, fresh concurrent close once.\n";
|