178 lines
14 KiB
PHP
178 lines
14 KiB
PHP
<?php
|
|
|
|
namespace Tests\Benchmarks;
|
|
|
|
use App\Models\User;
|
|
use Illuminate\Support\Facades\DB;
|
|
use Tests\Support\IdentityFixture;
|
|
|
|
require_once __DIR__.'/ModuleAssociationLoadTest.php';
|
|
|
|
class ModuleAssociationConcurrencyTest extends ModuleAssociationLoadTest
|
|
{
|
|
protected array $jobs = [];
|
|
protected string $directory;
|
|
protected int $serial = 0;
|
|
|
|
public function test_synthetic_volume_and_query_plans(): void
|
|
{
|
|
$run = (string) getenv('SGF_ISOLATED_REGRESSION_RUN');
|
|
$this->assertMatchesRegularExpression('/^[a-f0-9]{12}$/D', $run);
|
|
$identity = DB::selectOne("SELECT current_database() AS database,current_user AS role,inet_server_port() AS port,current_setting('cluster_name') AS cluster");
|
|
$this->assertSame('sgf_tempo_test', $identity->database); $this->assertSame('sgf_test_runner', $identity->role);
|
|
$this->assertSame('sgf_modular_regression_'.$run, $identity->cluster); $this->assertSame(55439, (int) $identity->port);
|
|
$tag = bin2hex(random_bytes(6)); $index = 'sgf_qa_read_revision_'.$tag;
|
|
$this->directory = base_path('../private/association-concurrency-'.$tag); mkdir($this->directory, 0700, true);
|
|
$before = $this->baseline();
|
|
$this->assertSame(0, (int) DB::selectOne("SELECT count(*) AS n FROM pg_indexes WHERE schemaname='public' AND tablename='identity_audit' AND indexname <> 'identity_audit_pkey'")->n, 'Unindexed baseline must be explicit');
|
|
$report = ['date' => '2026-11-02', 'run_tag' => $tag, 'identity' => $identity, 'optimized_sha256' => hash_file('sha256', base_path('app/Modules/Identity/Services/ModuleAssociationReview.php')),
|
|
'method' => 'Independent PHP processes and synthetic identities, pre-authenticated in-process HTTP kernels, synchronized starts, closed loop of five reads per worker, three waves; no external HTTP/login/browser capacity claim.',
|
|
'fixture_policy' => 'New synthetic scope committed only in regression DB for subprocess visibility; retained with private manifest. Existing rows preserved, not reset or deleted.', 'phases' => []];
|
|
$f = $this->fixture(3000, true); $scope = $f['identity'];
|
|
$fourth = User::create(['name' => 'Consulta carga sintética 4', 'email' => 'assoc-load-'.$tag.'@example.test', 'password' => bin2hex(random_bytes(24))]);
|
|
DB::table('company_memberships')->insert(['user_id' => $fourth->id, 'tenant_id' => $scope['tenant'], 'company_id' => $scope['company']]);
|
|
$actors = [$scope['manager']->id, $scope['reader']->id, $scope['limited']->id, $fourth->id];
|
|
foreach ($actors as $actor) {
|
|
foreach (['company.view', 'access.manage'] as $permission) IdentityFixture::grant($actor, $scope['tenant'], $scope['company'], $permission);
|
|
IdentityFixture::grant($actor, $scope['tenant'], $scope['company'], 'unit.view', $scope['unit']);
|
|
foreach (['clients', 'services', 'service_drafts', 'billing_drafts'] as $area) IdentityFixture::grant($actor, $scope['tenant'], $scope['company'], "$area.view", str_ends_with($area, '_drafts') ? $scope['unit'] : null);
|
|
}
|
|
IdentityFixture::grant($actors[0], $scope['tenant'], $scope['company'], 'clients.edit');
|
|
for ($i = 0; $i < 25; $i++) {
|
|
$lines = []; foreach (array_slice($f['ids']['services'], $i * 100, 100) as $id) $lines[] = ['service_id' => $id, 'quantity' => '1', 'unit_price' => '10', 'description' => str_repeat('x', 1000)];
|
|
$data = ['currency' => 'MZN', 'tax_status' => 'pending', 'accounting_status' => 'not_processed', 'lines' => $lines];
|
|
DB::table('a_service_drafts')->where('id', $f['ids']['service_drafts'][$i])->update(['version' => 2, 'data' => json_encode($data)]);
|
|
$data['source_version'] = 2; DB::table('a_billing_drafts')->where('id', $f['ids']['billing_drafts'][$i])->update(['version' => 2, 'data' => json_encode($data)]);
|
|
}
|
|
foreach (['a_service_drafts', 'a_billing_drafts', 'identity_audit'] as $table) DB::statement('ANALYZE '.$table);
|
|
$this->actingAs($scope['manager'])->withSession(['auth_version' => $scope['manager']->auth_version]);
|
|
$context = $this->postJson('/api/context', ['tenant_id' => $scope['tenant'], 'company_id' => $scope['company'], 'unit_id' => $scope['unit'], 'module_id' => null])->assertOk()->json('context.token');
|
|
$this->withHeader('X-SGF-Context', $context);
|
|
$workloads = [];
|
|
foreach (['clients', 'billing_drafts'] as $area) {
|
|
$url = '/api/module-records?area='.$area.'&per_page=25'; $r = $this->getJson($url)->assertOk()->assertJsonPath('pagination.total', 3000);
|
|
$workloads[$area] = ['url' => $url, 'total' => 3000, 'response_sha256' => hash('sha256', $r->getContent())];
|
|
}
|
|
// Separate processes cannot see uncommitted rows. Only this new QA scope is committed.
|
|
DB::commit();
|
|
file_put_contents($this->directory.'/manifest.json', json_encode(['synthetic_only' => true, 'tenant' => $scope['tenant'], 'company' => $scope['company'], 'unit' => $scope['unit'], 'actors' => $actors, 'workloads' => $workloads, 'index_name' => $index], JSON_PRETTY_PRINT | JSON_THROW_ON_ERROR));
|
|
$created = false;
|
|
try {
|
|
foreach (['none', 'revision'] as $mode) {
|
|
if ($mode === 'revision') {
|
|
$this->assertNull(DB::selectOne('SELECT to_regclass(?) AS value', [$index])->value);
|
|
DB::statement("CREATE INDEX $index ON identity_audit (tenant_id,company_id,(details->>'area'),(details->>'record_id'),id DESC) WHERE action IN ('module.record.associated','module.record.disassociated','module.record.reviewed')"); $created = true;
|
|
DB::statement('ANALYZE identity_audit');
|
|
}
|
|
$phase = ['index_mode' => $mode, 'waves' => []];
|
|
foreach ($workloads as $label => $workload) foreach ([1, 2, 4] as $count) for ($wave = 0; $wave < 3; $wave++) {
|
|
$phase['waves'][] = ['workload' => $label, 'configured_workers' => $count, ...$this->wave($scope, array_slice($actors, 0, $count), $workload, 5, $tag)];
|
|
fwrite(STDOUT, "\nCONCURRENT $mode $label workers=$count wave=".($wave + 1)." complete\n");
|
|
}
|
|
$report['phases'][] = $phase;
|
|
}
|
|
$candidate = array_values(array_filter($f['ids']['clients'], fn ($id) => $id % 3 === 0))[0];
|
|
$input = ['area' => 'clients', 'id' => $candidate, 'version' => 1, 'operation' => 'associate', 'module_id' => 'commercial', 'reason' => 'Synthetic concurrency rollback only'];
|
|
$first = $this->postJson('/api/module-records/preview', $input)->assertOk()->json('review_token');
|
|
$second = $this->postJson('/api/module-records/preview', $input)->assertOk()->json('review_token');
|
|
$events = DB::table('identity_audit')->where('tenant_id', $scope['tenant'])->where('company_id', $scope['company'])->where('action', 'like', 'module.record.%')->count();
|
|
DB::beginTransaction();
|
|
try {
|
|
$this->postJson('/api/module-records', ['review_token' => $first])->assertOk();
|
|
$report['mixed_read_wave'] = $this->wave($scope, $actors, $workloads['clients'], 5, $tag);
|
|
$gate = $this->directory.'/confirm-go';
|
|
$job = $this->launch(['mode' => 'confirm', 'actor' => $actors[0], 'tenant' => $scope['tenant'], 'company' => $scope['company'], 'unit' => $scope['unit'], 'context' => $context,
|
|
'review_token' => $second, 'go' => $gate, 'application_name' => 'sgf_assoc_confirm_'.$tag]);
|
|
$this->ready([$job]); file_put_contents($gate, 'go'); $observed = false; $deadline = microtime(true) + 3;
|
|
do { DB::select('SELECT pg_stat_clear_snapshot()'); $n = DB::selectOne("SELECT count(*) AS n FROM pg_stat_activity WHERE application_name=? AND wait_event_type='Lock'", ['sgf_assoc_confirm_'.$tag])->n; if ((int) $n > 0) { $observed = true; break; } usleep(25000); } while (microtime(true) < $deadline);
|
|
$result = $this->collect($job); $this->assertTrue($observed, 'Actual PostgreSQL lock wait must be observed');
|
|
$this->assertSame('55P03', $result['confirmation']['sqlstate'] ?? null);
|
|
$report['competing_confirmation'] = ['actual_lock_wait_observed' => $observed, ...$result['confirmation']];
|
|
} finally { DB::rollBack(); }
|
|
$this->assertSame($events, DB::table('identity_audit')->where('tenant_id', $scope['tenant'])->where('company_id', $scope['company'])->where('action', 'like', 'module.record.%')->count());
|
|
$report['synthetic_confirmation_rolled_back'] = true;
|
|
} finally {
|
|
foreach ($this->jobs as $job) if (is_resource($job['process'])) { proc_terminate($job['process']); proc_close($job['process']); }
|
|
while (DB::transactionLevel() > 0) DB::rollBack();
|
|
if ($created) DB::statement('DROP INDEX '.$index);
|
|
$this->assertNull(DB::selectOne('SELECT to_regclass(?) AS value', [$index])->value);
|
|
$report['test_index_removed'] = true;
|
|
$report['previous_rows_unchanged'] = $this->baseline($before) === $before;
|
|
$this->assertTrue($report['previous_rows_unchanged']);
|
|
$output = getenv('SGF_ASSOCIATION_LOAD_OUTPUT');
|
|
if (!$output || realpath(dirname($output)) !== realpath(base_path('..'))) throw new \RuntimeException('Unexpected report directory');
|
|
file_put_contents($output, json_encode($report, JSON_PRETTY_PRINT | JSON_THROW_ON_ERROR).PHP_EOL);
|
|
// Preserve the transaction-based TestCase lifecycle without undoing committed QA setup.
|
|
DB::beginTransaction();
|
|
}
|
|
}
|
|
|
|
private function wave(array $scope, array $actors, array $workload, int $requests, string $tag): array
|
|
{
|
|
$gate = $this->directory.'/go-'.(++$this->serial); $jobs = []; $startup = microtime(true);
|
|
foreach ($actors as $i => $actor) $jobs[] = $this->launch(['mode' => 'read', 'actor' => $actor, 'tenant' => $scope['tenant'], 'company' => $scope['company'], 'unit' => $scope['unit'],
|
|
...$workload, 'requests' => $requests, 'go' => $gate, 'application_name' => 'sgf_assoc_read_'.$tag.'_'.$i]);
|
|
$this->ready($jobs); $ready = microtime(true); file_put_contents($gate, 'go');
|
|
$samples = []; $finished = $ready;
|
|
foreach ($jobs as $job) { $result = $this->collect($job); $samples = [...$samples, ...$result['requests']]; $finished = max($finished, $result['finished_at']); }
|
|
$ok = array_values(array_filter($samples, fn ($s) => $s['status'] === 'ok')); $events = [];
|
|
foreach ($ok as $s) { $events[] = [$s['started_at'], 1]; $events[] = [$s['finished_at'], -1]; }
|
|
usort($events, fn ($a, $b) => ($a[0] <=> $b[0]) ?: ($a[1] <=> $b[1])); $active = 0; $max = 0;
|
|
foreach ($events as $event) { $active += $event[1]; $max = max($max, $active); }
|
|
$this->assertLessThanOrEqual(count($actors), $max); $this->assertGreaterThanOrEqual(1, $max);
|
|
return ['startup_to_ready_ms' => round(($ready - $startup) * 1000, 3), 'measurement_window_ms' => round(($finished - $ready) * 1000, 3), 'observed_max_inflight' => $max,
|
|
'complete_requests' => count($ok), 'limited_requests' => count($samples) - count($ok), 'samples' => $samples];
|
|
}
|
|
|
|
protected function launch(array $input): array
|
|
{
|
|
$stem = $this->directory.'/worker-'.(++$this->serial);
|
|
$input += ['ready' => $stem.'.ready', 'result' => $stem.'.json'];
|
|
file_put_contents($stem.'.input.json', json_encode($input, JSON_THROW_ON_ERROR));
|
|
$env = getenv(); $env['SGF_ASSOCIATION_WORKER_INPUT'] = $stem.'.input.json';
|
|
$pipes = []; $process = proc_open([PHP_BINARY, 'vendor/phpunit/phpunit/phpunit', 'scripts/benchmarks/ModuleAssociationReadWorkerTest.php', '--no-progress'],
|
|
[0 => ['pipe', 'r'], 1 => ['file', $stem.'.stdout', 'w'], 2 => ['file', $stem.'.stderr', 'w']], $pipes, base_path(), $env, ['bypass_shell' => true]);
|
|
$this->assertIsResource($process); fclose($pipes[0]);
|
|
$job = ['process' => $process, 'input' => $input, 'stdout' => $stem.'.stdout', 'stderr' => $stem.'.stderr']; $this->jobs[] = $job; return $job;
|
|
}
|
|
|
|
protected function ready(array $jobs): void
|
|
{
|
|
$deadline = microtime(true) + 30;
|
|
do {
|
|
$all = true;
|
|
foreach ($jobs as $job) if (!is_file($job['input']['ready'])) { $all = false; if (!proc_get_status($job['process'])['running']) throw new \RuntimeException('Worker failed before ready: '.file_get_contents($job['stdout']).file_get_contents($job['stderr'])); }
|
|
if ($all) return; usleep(10000);
|
|
} while (microtime(true) < $deadline);
|
|
throw new \RuntimeException('Workers did not reach barrier');
|
|
}
|
|
|
|
protected function collect(array $job): array
|
|
{
|
|
$deadline = microtime(true) + 60;
|
|
do { $status = proc_get_status($job['process']); if (!$status['running']) break; usleep(10000); } while (microtime(true) < $deadline);
|
|
if ($status['running']) throw new \RuntimeException('Worker deadline reached');
|
|
$exit = proc_close($job['process']); if ($exit === -1) $exit = $status['exitcode'];
|
|
$this->assertSame(0, $exit, file_get_contents($job['stdout']).file_get_contents($job['stderr']));
|
|
return json_decode(file_get_contents($job['input']['result']), true, 512, JSON_THROW_ON_ERROR);
|
|
}
|
|
|
|
protected function baseline(?array $old = null): array
|
|
{
|
|
$out = [];
|
|
foreach (['a_clients', 'a_services', 'a_service_drafts', 'a_billing_drafts', 'identity_audit', 'users', 'permission_grants', 'company_memberships'] as $table) {
|
|
$key = $table === 'company_memberships' ? 'user_id' : 'id';
|
|
$q = DB::table($table);
|
|
// IDs are captured integer primary keys, not user input. Avoid the
|
|
// PostgreSQL bind-parameter ceiling in accumulated QA fixtures.
|
|
if ($old !== null) $q->whereIntegerInRaw($key, $old[$table]['ids']);
|
|
$q->orderBy($key);
|
|
if ($table === 'company_memberships') $q->orderBy('tenant_id')->orderBy('company_id');
|
|
$rows = $q->get();
|
|
$out[$table] = ['ids' => $old[$table]['ids'] ?? $rows->pluck($key)->unique()->values()->all(), 'hash' => hash('sha256', $rows->toJson())];
|
|
}
|
|
$out['indexes'] = json_encode(DB::select("SELECT indexname,indexdef FROM pg_indexes WHERE schemaname='public' ORDER BY indexname"), JSON_THROW_ON_ERROR);
|
|
return $out;
|
|
}
|
|
}
|