Extract WorkerRunner from WorkerCommand

Moves everything WorkerCommand does after CommandHelper::begin() — the
TCP connect, NDJSON handshake and per-file analysis loop — into a
reusable WorkerRunner service. WorkerCommand now only boots and
delegates. This makes the post-boot worker logic callable without a
re-boot, which the pcntl_fork() path will rely on.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Author: ondrejmirtes

src/Command/WorkerCommand.php

@@ -2,47 +2,25 @@
 
 namespace PHPStan\Command;
 
-use Clue\React\NDJson\Decoder;
-use Clue\React\NDJson\Encoder;
 use Override;
-use PHPStan\Analyser\FileAnalyser;
-use PHPStan\Analyser\InternalError;
-use PHPStan\Analyser\NodeScopeResolver;
-use PHPStan\Collectors\Registry as CollectorRegistry;
-use PHPStan\DependencyInjection\Container;
 use PHPStan\File\PathNotFoundException;
-use PHPStan\Rules\Registry as RuleRegistry;
+use PHPStan\Parallel\WorkerRunner;
 use PHPStan\ShouldNotHappenException;
-use React\EventLoop\StreamSelectLoop;
-use React\Socket\ConnectionInterface;
-use React\Socket\TcpConnector;
-use React\Stream\ReadableStreamInterface;
-use React\Stream\WritableStreamInterface;
 use Symfony\Component\Console\Command\Command;
 use Symfony\Component\Console\Input\InputArgument;
 use Symfony\Component\Console\Input\InputInterface;
 use Symfony\Component\Console\Input\InputOption;
 use Symfony\Component\Console\Output\OutputInterface;
-use Throwable;
-use function array_fill_keys;
-use function array_filter;
-use function array_merge;
-use function array_unshift;
-use function array_values;
-use function defined;
 use function is_array;
 use function is_bool;
 use function is_string;
-use function memory_get_peak_usage;
 use function sprintf;
 
 final class WorkerCommand extends Command
 {
 
 	private const NAME = 'worker';
 
-	private int $errorCount = 0;
-
 	/**
 	 * @param string[] $composerAutoloaderProjectPaths
 	 */
@@ -122,196 +100,30 @@ protected function execute(InputInterface $input, OutputInterface $output): int
 		} catch (InceptionNotSuccessfulException $e) {
 			return 1;
 		}
-		$loop = new StreamSelectLoop();
 
 		$container = $inceptionResult->getContainer();
 
 		try {
 			[$analysedFiles] = $inceptionResult->getFiles();
-			$analysedFiles = $this->switchTmpFile($analysedFiles, $insteadOfFile, $tmpFile);
 		} catch (PathNotFoundException $e) {
 			$inceptionResult->getErrorOutput()->writeLineFormatted(sprintf('<error>%s</error>', $e->getMessage()));
 			return 1;
 		} catch (InceptionNotSuccessfulException) {
 			return 1;
 		}
 
-		$nodeScopeResolver = $container->getByType(NodeScopeResolver::class);
-		$nodeScopeResolver->setAnalysedFiles($analysedFiles);
-
-		$analysedFiles = array_fill_keys($analysedFiles, true);
-
-		$tcpConnector = new TcpConnector($loop);
-		$tcpConnector->connect(sprintf('127.0.0.1:%d', (int) $port))->then(function (ConnectionInterface $connection) use ($container, $identifier, $output, $analysedFiles, $tmpFile, $insteadOfFile): void {
-			// phpcs:disable SlevomatCodingStandard.Namespaces.ReferenceUsedNamesOnly
-			$jsonInvalidUtf8Ignore = defined('JSON_INVALID_UTF8_IGNORE') ? JSON_INVALID_UTF8_IGNORE : 0;
-			// phpcs:enable
-			$out = new Encoder($connection, $jsonInvalidUtf8Ignore);
-			$in = new Decoder($connection, true, options: $jsonInvalidUtf8Ignore, maxlength: $container->getParameter('parallel')['buffer']);
-			$out->write(['action' => 'hello', 'identifier' => $identifier]);
-			$this->runWorker($container, $out, $in, $output, $analysedFiles, $tmpFile, $insteadOfFile);
-		});
-
-		$loop->run();
-
-		if ($this->errorCount > 0) {
-			return 1;
-		}
-
-		return 0;
-	}
-
-	/**
-	 * @param array<string, true> $analysedFiles
-	 */
-	private function runWorker(
-		Container $container,
-		WritableStreamInterface $out,
-		ReadableStreamInterface $in,
-		OutputInterface $output,
-		array $analysedFiles,
-		?string $tmpFile,
-		?string $insteadOfFile,
-	): void
-	{
-		$handleError = function (Throwable $error) use ($out, $output): void {
-			$this->errorCount++;
-			$output->writeln(sprintf('Error: %s', $error->getMessage()));
-			$out->write([
-				'action' => 'result',
-				'result' => [
-					'errors' => [],
-					'internalErrors' => [
-						new InternalError(
-							$error->getMessage(),
-							'communicating with main process in parallel worker',
-							InternalError::prepareTrace($error),
-							$error->getTraceAsString(),
-							shouldReportBug: true,
-						),
-					],
-					'filteredPhpErrors' => [],
-					'allPhpErrors' => [],
-					'locallyIgnoredErrors' => [],
-					'linesToIgnore' => [],
-					'unmatchedLineIgnores' => [],
-					'collectedData' => [],
-					'memoryUsage' => memory_get_peak_usage(true),
-					'dependencies' => [],
-					'exportedNodes' => [],
-					'files' => [],
-					'internalErrorsCount' => 1,
-				],
-			]);
-			$out->end();
-		};
-		$out->on('error', $handleError);
-		$fileAnalyser = $container->getByType(FileAnalyser::class);
-		$ruleRegistry = $container->getByType(RuleRegistry::class);
-		$collectorRegistry = $container->getByType(CollectorRegistry::class);
-		$in->on('data', static function (array $json) use ($fileAnalyser, $ruleRegistry, $collectorRegistry, $out, $analysedFiles, $tmpFile, $insteadOfFile): void {
-			$action = $json['action'];
-			if ($action !== 'analyse') {
-				return;
-			}
-
-			$internalErrorsCount = 0;
-			$files = $json['files'];
-			$errors = [];
-			$internalErrors = [];
-			$filteredPhpErrors = [];
-			$allPhpErrors = [];
-			$locallyIgnoredErrors = [];
-			$linesToIgnore = [];
-			$unmatchedLineIgnores = [];
-			$collectedData = [];
-			$dependencies = [];
-			$usedTraitDependencies = [];
-			$exportedNodes = [];
-			$processedFiles = [];
-			foreach ($files as $file) {
-				try {
-					if ($file === $insteadOfFile) {
-						$file = $tmpFile;
-					}
-					$fileAnalyserResult = $fileAnalyser->analyseFile($file, $analysedFiles, $ruleRegistry, $collectorRegistry, null);
-					$fileErrors = $fileAnalyserResult->getErrors();
-					$filteredPhpErrors = array_merge($filteredPhpErrors, $fileAnalyserResult->getFilteredPhpErrors());
-					$allPhpErrors = array_merge($allPhpErrors, $fileAnalyserResult->getAllPhpErrors());
-					$linesToIgnore[$file] = $fileAnalyserResult->getLinesToIgnore();
-					$unmatchedLineIgnores[$file] = $fileAnalyserResult->getUnmatchedLineIgnores();
-					$dependencies[$file] = $fileAnalyserResult->getDependencies();
-					$usedTraitDependencies[$file] = $fileAnalyserResult->getUsedTraitDependencies();
-					$exportedNodes[$file] = $fileAnalyserResult->getExportedNodes();
-					$processedFiles = array_merge($processedFiles, $fileAnalyserResult->getProcessedFiles());
-					foreach ($fileErrors as $fileError) {
-						$errors[] = $fileError;
-					}
-					foreach ($fileAnalyserResult->getLocallyIgnoredErrors() as $locallyIgnoredError) {
-						$locallyIgnoredErrors[] = $locallyIgnoredError;
-					}
-					foreach ($fileAnalyserResult->getCollectedData() as $collectedFile => $dataPerCollector) {
-						foreach ($dataPerCollector as $collectorType => $collectorData) {
-							foreach ($collectorData as $data) {
-								$collectedData[$collectedFile][$collectorType][] = $data;
-							}
-						}
-					}
-				} catch (Throwable $t) {
-					$internalErrorsCount++;
-					$internalErrors[] = new InternalError(
-						$t->getMessage(),
-						sprintf('analysing file %s', $file),
-						InternalError::prepareTrace($t),
-						$t->getTraceAsString(),
-						shouldReportBug: true,
-					);
-				}
-			}
-
-			$out->write([
-				'action' => 'result',
-				'result' => [
-					'errors' => $errors,
-					'internalErrors' => $internalErrors,
-					'filteredPhpErrors' => $filteredPhpErrors,
-					'allPhpErrors' => $allPhpErrors,
-					'locallyIgnoredErrors' => $locallyIgnoredErrors,
-					'linesToIgnore' => $linesToIgnore,
-					'unmatchedLineIgnores' => $unmatchedLineIgnores,
-					'collectedData' => $collectedData,
-					'memoryUsage' => memory_get_peak_usage(true),
-					'dependencies' => $dependencies,
-					'usedTraitDependencies' => $usedTraitDependencies,
-					'exportedNodes' => $exportedNodes,
-					'files' => $files,
-					'processedFiles' => $processedFiles,
-					'internalErrorsCount' => $internalErrorsCount,
-				]]);
-		});
-		$in->on('error', $handleError);
-	}
-
-	/**
-	 * @param string[] $analysedFiles
-	 * @return string[]
-	 */
-	private function switchTmpFile(
-		array $analysedFiles,
-		?string $insteadOfFile,
-		?string $tmpFile,
-	): array
-	{
-		if ($insteadOfFile === null) {
-			return $analysedFiles;
-		}
-		$analysedFiles = array_values(array_filter($analysedFiles, static fn (string $file): bool => $file !== $insteadOfFile));
-
-		if ($tmpFile !== null) {
-			array_unshift($analysedFiles, $tmpFile);
-		}
-
-		return $analysedFiles;
+		// Everything after the boot lives in WorkerRunner so a pcntl_fork()-ed
+		// child can reuse it without re-booting (see ParallelAnalyser).
+		$workerRunner = $container->getByType(WorkerRunner::class);
+
+		return $workerRunner->run(
+			$output,
+			$analysedFiles,
+			(int) $port,
+			$identifier,
+			$tmpFile,
+			$insteadOfFile,
+		);
 	}
 
 }