152 lines
9.0 KiB
PHP
152 lines
9.0 KiB
PHP
<?php
|
|
|
|
declare(strict_types=1);
|
|
|
|
require dirname(__DIR__) . '/vendor/autoload.php';
|
|
|
|
use app\adminapi\service\AssistantRequestControl;
|
|
|
|
function stopExpect(bool $ok, string $message): void {
|
|
if (!$ok) throw new RuntimeException($message);
|
|
}
|
|
|
|
if (($argv[1] ?? '') === 'worker') {
|
|
[$unused, $workerFlag, $mode, $directory, $id] = $argv;
|
|
$control = new AssistantRequestControl(7, 11, $id, $directory);
|
|
$deadline = microtime(true) + 5;
|
|
while (!is_file($directory . '/gate') && microtime(true) < $deadline) usleep(1000);
|
|
echo json_encode($control->requestCancel(static function (array $task) use ($directory, $mode): bool {
|
|
file_put_contents($directory . '/calls', $task['task_id'] . "\n", FILE_APPEND | LOCK_EX);
|
|
if ($mode === 'abandon') exit(0); // Simulate a worker dying with its lease still held.
|
|
usleep(180000);
|
|
return true;
|
|
}));
|
|
exit;
|
|
}
|
|
|
|
$directory = sys_get_temp_dir() . '/zyt-stop-coordination-' . bin2hex(random_bytes(8));
|
|
$make = static function (int $number) use ($directory): AssistantRequestControl {
|
|
$control = new AssistantRequestControl(7, 11, sprintf('12345678-1234-4234-8234-%012d', $number), $directory);
|
|
stopExpect($control->begin(), 'fresh control begins');
|
|
$control->bindTask('qwen', 'synthetic-stop-user', 'synthetic-stop-' . $number);
|
|
return $control;
|
|
};
|
|
$launch = static function (string $mode, int $number) use ($directory): array {
|
|
$pipes = [];
|
|
$process = proc_open([PHP_BINARY, __FILE__, 'worker', $mode, $directory,
|
|
sprintf('12345678-1234-4234-8234-%012d', $number)],
|
|
[0 => ['pipe', 'r'], 1 => ['pipe', 'w'], 2 => ['pipe', 'w']], $pipes);
|
|
stopExpect(is_resource($process), 'independent stop worker launches');
|
|
fclose($pipes[0]);
|
|
return [$process, $pipes];
|
|
};
|
|
$join = static function (array $worker): string {
|
|
[$process, $pipes] = $worker;
|
|
$stdout = stream_get_contents($pipes[1]); $stderr = stream_get_contents($pipes[2]);
|
|
fclose($pipes[1]); fclose($pipes[2]);
|
|
stopExpect(proc_close($process) === 0 && $stderr === '', 'independent stop worker exits cleanly');
|
|
return $stdout;
|
|
};
|
|
try {
|
|
$control = $make(1);
|
|
$workers = [];
|
|
for ($index = 0; $index < 4; ++$index) $workers[] = $launch('race', 1);
|
|
file_put_contents($directory . '/gate', '1');
|
|
foreach ($workers as $worker) $join($worker);
|
|
$calls = file($directory . '/calls', FILE_IGNORE_NEW_LINES);
|
|
stopExpect(count($calls) === 1 && $control->publicState()['upstream_stop_confirmed'],
|
|
'four independent workers send exactly one accepted stop for the same task');
|
|
stopExpect(!$control->publicState()['upstream_terminal_confirmed'], 'stop API acceptance does not imply workflow termination');
|
|
$ownerCalls = 0;
|
|
$ownerOutcome = $control->stopBoundTask(static function () use (&$ownerCalls): bool { ++$ownerCalls; return true; }, 'synthetic-stop-1');
|
|
stopExpect($ownerOutcome['accepted'] && $ownerCalls === 0, 'owner observes peer acceptance without a second HTTP stop');
|
|
|
|
$abandoned = $make(2);
|
|
$join($launch('abandon', 2));
|
|
$takeovers = 0;
|
|
$compensate = static function () use (&$takeovers): bool { ++$takeovers; return true; };
|
|
stopExpect($abandoned->stopBoundTask($compensate)['pending'] && $takeovers === 0,
|
|
'an active claim lease blocks a duplicate stop');
|
|
usleep(1300000);
|
|
stopExpect($abandoned->stopBoundTask($compensate)['accepted'] && $takeovers === 1,
|
|
'an abandoned lease expires and allows one bounded takeover');
|
|
|
|
$failed = $make(3); $failed->signalCancel();
|
|
$attempts = 0;
|
|
$reject = static function () use (&$attempts): bool { ++$attempts; return false; };
|
|
stopExpect(!$failed->stopBoundTask($reject)['accepted'], 'a rejected stop remains unconfirmed');
|
|
$failed->clearTask('synthetic-stop-3'); $failed->finish('cancelled');
|
|
usleep(260000);
|
|
$retryState = $failed->requestCancel(static function () use (&$attempts): bool { ++$attempts; return true; });
|
|
stopExpect($attempts === 2 && $retryState['upstream_stop_confirmed'] && $retryState['transport_closed'],
|
|
'failed known task remains available for compensation after local transport termination');
|
|
$failed->requestCancel($reject);
|
|
stopExpect($attempts === 2, 'accepted compensation stays idempotent');
|
|
|
|
$bounded = $make(4); $bounded->signalCancel(); $attempts = 0;
|
|
$bounded->stopBoundTask($reject); usleep(260000); $bounded->stopBoundTask($reject);
|
|
usleep(260000); $bounded->requestCancel($reject);
|
|
stopExpect($attempts === 2 && !$bounded->publicState()['upstream_stop_confirmed'], 'failure never creates an unbounded stop retry loop');
|
|
|
|
$expired = $make(5); $expired->signalCancel();
|
|
$path = $directory . '/' . hash('sha256', '7:11:12345678-1234-4234-8234-000000000005') . '.json';
|
|
$state = json_decode((string) file_get_contents($path), true);
|
|
$state['stop_retry_until'] = microtime(true) - 1;
|
|
file_put_contents($path, json_encode($state));
|
|
$attempts = 0; $expired->requestCancel($reject);
|
|
stopExpect($attempts === 0, 'expired compensation window is not extended by repeated cancel requests');
|
|
|
|
$generation = $make(6); $generation->signalValidationFailure();
|
|
stopExpect(!$generation->canStartAttempt(), 'unconfirmed old task prevents another generation');
|
|
try {
|
|
$generation->bindTask('qwen', 'synthetic-stop-user', 'synthetic-race-task');
|
|
stopExpect(false, 'a task handoff cannot overwrite the unconfirmed task');
|
|
} catch (RuntimeException $e) {
|
|
stopExpect($e->getMessage() === 'UPSTREAM_TASK_HANDOFF_BLOCKED', 'unexpected incoming task is quarantined');
|
|
}
|
|
stopExpect($generation->hasHandoffTask('synthetic-race-task') && !$generation->shouldAbort(),
|
|
'unexpected task is retained without inventing user cancellation');
|
|
$generation->stopBoundTask(static fn (): bool => true, 'synthetic-race-task');
|
|
$generation->markTerminal('synthetic-race-task', 'stopped');
|
|
$generation->clearTask('synthetic-race-task');
|
|
stopExpect(!$generation->canStartAttempt(), 'terminating the race task does not terminate the original task');
|
|
$generation->markStop(true, 'synthetic-stop-6');
|
|
stopExpect(!$generation->canStartAttempt(), 'stop acceptance alone cannot authorize repair');
|
|
$generation->markTerminal('synthetic-stop-6', 'stopped');
|
|
$generation->clearTask('synthetic-stop-6');
|
|
stopExpect($generation->canStartAttempt(), 'confirmed terminal permits the next private generation');
|
|
$generation->bindTask('qwen', 'synthetic-stop-user', 'synthetic-next-generation');
|
|
$generation->markStop(true, 'synthetic-stop-6');
|
|
$generation->markTerminal('synthetic-stop-6', 'stopped');
|
|
$generation->clearTask('synthetic-stop-6');
|
|
stopExpect(!$generation->publicState()['upstream_stop_confirmed']
|
|
&& !$generation->publicState()['upstream_terminal_confirmed'], 'late stop and terminal callbacks cannot poison the next task generation');
|
|
$attempts = 0;
|
|
$generation->stopBoundTask($reject, 'synthetic-stop-6');
|
|
stopExpect($attempts === 0, 'a stale owner cannot stop a different task generation');
|
|
$generation->signalCancel();
|
|
stopExpect(!$generation->canStartAttempt(), 'real user cancellation blocks new attempts');
|
|
stopExpect($generation->stopBoundTask(static fn (): bool => true, 'synthetic-next-generation')['accepted'],
|
|
'current task generation remains independently stoppable');
|
|
$generation->markTerminal('synthetic-next-generation', 'running');
|
|
stopExpect(!$generation->publicState()['upstream_terminal_confirmed'], 'running metadata is never accepted as terminal');
|
|
$generation->markTerminal('synthetic-next-generation', 'stopped');
|
|
$generation->markTerminal('synthetic-next-generation', 'failed');
|
|
stopExpect($generation->publicState()['upstream_terminal_confirmed']
|
|
&& $generation->publicState()['upstream_terminal_status'] === 'stopped', 'only matching terminal metadata confirms the workflow state');
|
|
$sibling = $make(7);
|
|
stopExpect(!$sibling->shouldAbort() && !$sibling->publicState()['upstream_stop_confirmed'], 'new request UUID is isolated from all old task callbacks');
|
|
stopExpect(!str_contains(json_encode($generation->publicState()), 'synthetic-next-generation'), 'public state never discloses task identifiers');
|
|
$legacy = $make(8); $legacy->signalCancel();
|
|
$legacyPath = $directory . '/' . hash('sha256', '7:11:12345678-1234-4234-8234-000000000008') . '.json';
|
|
$legacyState = json_decode((string) file_get_contents($legacyPath), true);
|
|
unset($legacyState['upstream']['generation']);
|
|
file_put_contents($legacyPath, json_encode($legacyState));
|
|
stopExpect($legacy->stopBoundTask(static fn (): bool => true)['accepted'], 'in-flight task state from the previous release is upgraded before claiming');
|
|
echo "Diagnosis AI stop claims / multi-process dedup / lease recovery / compensation / task generations: OK\n";
|
|
} finally {
|
|
foreach (glob($directory . '/*') ?: [] as $path) unlink($path);
|
|
if (is_file($directory . '/.gc.lock')) unlink($directory . '/.gc.lock');
|
|
if (is_dir($directory)) rmdir($directory);
|
|
}
|