75 lines
4.0 KiB
PHP
75 lines
4.0 KiB
PHP
<?php
|
|
|
|
use Illuminate\Support\Facades\DB;
|
|
use Symfony\Component\Process\Process;
|
|
|
|
/** Runs only inside the existing disposable-database proof, never on an application. */
|
|
function inboxPgRaces(array $environment, array $config, string $database): array
|
|
{
|
|
if (! preg_match('/^compras_receipts_test_[a-f0-9]{12}$/D', $database) || ($environment['INBOX_TEST_PG'] ?? null) !== '1') {
|
|
throw new LogicException('Disputas de recepção exigem destino descartável.');
|
|
}
|
|
config(['database.connections.inbox_proof_read' => [...$config, 'database' => $database, 'url' => null]]);
|
|
$reader = DB::connection('inbox_proof_read');
|
|
$worker = fn (string $mode, int $id = 1) => new Process([PHP_BINARY, 'tests/Concurrency/inbox-worker.php', $mode, (string) $id], dirname(__DIR__, 2), $environment, timeout: 20);
|
|
$races = [];
|
|
try {
|
|
foreach (['same_page', 'different_pages'] as $scenario) {
|
|
$worker('prepare')->mustRun();
|
|
$first = $worker('hold');
|
|
$first->start();
|
|
$deadline = microtime(true) + 5;
|
|
while ($first->isRunning() && ! str_contains($first->getOutput(), 'locked') && microtime(true) < $deadline) {
|
|
usleep(20000);
|
|
}
|
|
if (! str_contains($first->getOutput(), 'locked')) {
|
|
throw new RuntimeException('Primeiro receptor sem bloqueio demonstrado.');
|
|
}
|
|
$second = $worker('run', $scenario === 'same_page' ? 1 : 2);
|
|
$second->start();
|
|
$blocked = false;
|
|
$deadline = microtime(true) + 5;
|
|
while ($first->isRunning() && microtime(true) < $deadline) {
|
|
$blocked = (bool) $reader->selectOne("SELECT EXISTS (SELECT 1 FROM pg_stat_activity WHERE datname=? AND wait_event_type='Lock') AS blocked", [$database])->blocked;
|
|
if ($blocked) {
|
|
break;
|
|
}
|
|
usleep(20000);
|
|
}
|
|
$first->wait();
|
|
$second->wait();
|
|
if (! $first->isSuccessful() || ! $second->isSuccessful() || ! $blocked) {
|
|
throw new RuntimeException('Disputa sem execução/bloqueio confirmados.');
|
|
}
|
|
$a = json_decode(str_replace('locked', '', $first->getOutput()), true, 16, JSON_THROW_ON_ERROR);
|
|
$b = json_decode($second->getOutput(), true, 16, JSON_THROW_ON_ERROR);
|
|
$accepted = ($a['result'] === 'accepted' ? 1 : 0) + ($b['result'] === 'accepted' ? 1 : 0);
|
|
if ($accepted !== ($scenario === 'same_page' ? 2 : 1)
|
|
|| ($scenario === 'same_page' && (($a['inserted'] + $b['inserted']) !== 1 || ($a['replayed'] === $b['replayed'])))
|
|
|| (int) $reader->table('sgc_exchange_facts')->count() !== 1
|
|
|| (int) $reader->table('sgc_exchange_streams')->value('version') !== 2) {
|
|
throw new RuntimeException('Disputa duplicou factos ou fez avançar o cursor indevidamente.');
|
|
}
|
|
$races[] = ['scenario' => $scenario, 'real_lock_contention' => true, 'facts' => 1, 'checkpoint_version' => 2, 'result' => 'passed'];
|
|
}
|
|
$worker('prepare')->mustRun();
|
|
$interrupted = $worker('exit-after-commit');
|
|
$interrupted->run();
|
|
if ($interrupted->getExitCode() !== 42) {
|
|
throw new RuntimeException('Interrupção após commit não demonstrada.');
|
|
}
|
|
$retry = $worker('run');
|
|
$retry->mustRun();
|
|
$result = json_decode($retry->getOutput(), true, 16, JSON_THROW_ON_ERROR);
|
|
if (($result['replayed'] ?? false) !== true || $result['inserted'] !== 0 || (int) $reader->table('sgc_exchange_facts')->count() !== 1
|
|
|| (int) $reader->table('sgc_exchange_streams')->value('version') !== 2) {
|
|
throw new RuntimeException('Retry após perda da confirmação não foi idempotente.');
|
|
}
|
|
$races[] = ['scenario' => 'process_exit_after_commit_before_acknowledgement', 'result' => 'passed', 'facts' => 1, 'checkpoint_version' => 2];
|
|
} finally {
|
|
DB::purge('inbox_proof_read');
|
|
}
|
|
|
|
return $races;
|
|
}
|