#367 - process_launcher with json_buffering: output nothing for an empty buffer, remove the buffers of failed sub-processes

CHANGELOG.md

@@ -1,6 +1,9 @@
 Latest
 ------
 
+## Fixes
+* [#367](https://github.com/cleverage/process-bundle/issues/367) `ProcessLauncherTask` (`process_launcher`) with `json_buffering`: when the sub-process writes nothing into its JSON buffer (no `end_point`, or an output that is not an array), the task outputs nothing again, as before 5.5, instead of the path of an empty file, and the empty buffer is removed. The buffer of a failed sub-process, and those of the still running sub-processes stopped because of it, are removed too (one file was left in the temporary directory for each launch). Update documentation, add tests.
+
 v5.5
 -----
 

docs/04-advanced_workflow.md

@@ -316,7 +316,8 @@ PHP executes a process in a single thread. To use several CPU cores, the
   of its `end_point` is written to a JSON stream file whose path is outputted by the task once the sub-process is over
   (use a [JsonStreamReaderTask](reference/tasks/json_stream_reader_task.md) to read it back). The file is only written
   if the sub-process has an `end_point` whose last output is an array; otherwise, as when `json_buffering` is disabled,
-  the task outputs nothing
+  the task outputs nothing (and removes the empty file). The task does not remove the files it outputs: remove them
+  once read
 - if a sub-process fails (non-zero exit code), all the running sub-processes are stopped and the task throws an
   exception
 

docs/reference/tasks/process_launcher_task.md

@@ -26,8 +26,9 @@ Possible outputs
 
 Nothing is output when an input is received (the task is skipped). When `json_buffering` is enabled, the sub-process
 end point output (only if it is an array) is written to a JSON stream file of the system temporary directory (`<sys_get_temp_dir()>/cdm_buffer_*.json-stream`, in `var/` of the project before 5.4) and, once the sub-process is
-finished, the task outputs the path of this file (`string`), during the next iterations or when flushed. Without
-`json_buffering`, the task outputs nothing.
+finished, the task outputs the path of this file (`string`), during the next iterations or when flushed. If nothing
+was written (no `end_point`, or an output that is not an array), the empty file is removed and the task outputs
+nothing for this sub-process. Without `json_buffering`, the task outputs nothing.
 
 If a sub-process exits with a non-zero code, its error output is logged as critical, all running sub-processes are
 stopped and a `RuntimeException` is thrown.
@@ -69,5 +70,7 @@ Notes
 
 * Sub-processes are run with the console of the project (`%kernel.project_dir%/bin/console`) and the same Symfony
   environment as the current one (`--env`), the arguments of the service.
-* The JSON stream files are not removed by the task: remove them once read (e.g. with a [file remover](file_remover_task.md) task).
+* The JSON stream files output by the task are not removed by it: remove them once read (e.g. with a
+  [file remover](file_remover_task.md) task). The task only removes the files it does not output: empty ones, and
+  those of a failed sub-process and of the still running sub-processes stopped because of it.
 * The incremental error output (stderr) of the running sub-processes is echoed directly.

src/Model/SubprocessInstance.php

@@ -175,14 +175,36 @@ public function getContext(): array
         return $this->context;
     }
 
+    /**
+     * Path of the JSON buffer once the sub-process is terminated, null if nothing was written into it (no end point,
+     * output that is not an array): the empty buffer is then removed.
+     */
     public function getResult(): ?string
     {
         $fs = new Filesystem();
-        if ($this->process->isTerminated() && $fs->exists($this->bufferPath)) {
-            return $this->bufferPath;
+        if (!$this->process->isTerminated() || !$fs->exists($this->bufferPath)) {
+            return null;
+        }
+
+        // Created before the launch, the buffer exists even when the sub-process wrote nothing into it
+        clearstatcache(true, $this->bufferPath);
+        if (0 === filesize($this->bufferPath)) {
+            $this->removeBuffer();
+
+            return null;
         }
 
-        return null;
+        return $this->bufferPath;
+    }
+
+    /**
+     * Remove the JSON buffer, e.g. when the sub-process has failed or was stopped: it is never output.
+     */
+    public function removeBuffer(): void
+    {
+        if ('' !== $this->bufferPath) {
+            (new Filesystem())->remove($this->bufferPath);
+        }
     }
 
     /**

src/Task/Process/ProcessLauncherTask.php

@@ -202,6 +202,7 @@ protected function handleProcesses(ProcessState $state): void
 
             unset($this->launchedProcesses[$key]);
             if (0 !== $process->getProcess()->getExitCode()) {
+                $process->removeBuffer();
                 $this->logger->critical($process->getProcess()->getErrorOutput(), $logContext);
                 $this->killProcesses();
 
@@ -272,7 +273,12 @@ static function (Options $options, array $value): array {
     protected function killProcesses(): void
     {
         foreach ($this->launchedProcesses as $process) {
+            // An already terminated sub-process keeps its buffer: it is handled normally at the next call
+            $running = !$process->getProcess()->isTerminated();
             $process->stop(5);
+            if ($running) {
+                $process->removeBuffer();
+            }
         }
     }
 }

tests/Model/SubprocessInstanceTest.php

@@ -75,6 +75,49 @@ public function testJsonBufferIsWrittenInTheTemporaryDirectory(): void
         unlink($bufferPath);
     }
 
+    public function testEmptyJsonBufferIsRemovedAndNotReturned(): void
+    {
+        // The console outputs nothing: no end point, or an output that is not an array
+        $subprocess = $this->runWithJsonBuffering();
+        $bufferPath = $this->getBufferPath($subprocess);
+
+        $this->assertNull($subprocess->getResult());
+        $this->assertFileDoesNotExist($bufferPath);
+    }
+
+    public function testWrittenJsonBufferIsReturned(): void
+    {
+        file_put_contents($this->projectDir.'/bin/console', '<?php file_put_contents(substr($argv[array_key_last($argv) - 2], 9), "[\"a\"]\n");');
+        $subprocess = $this->runWithJsonBuffering();
+        $bufferPath = $this->getBufferPath($subprocess);
+
+        $this->assertSame($bufferPath, $subprocess->getResult());
+        $this->assertStringEqualsFile($bufferPath, "[\"a\"]\n");
+        unlink($bufferPath);
+    }
+
+    public function testNoResultWithoutJsonBuffering(): void
+    {
+        $subprocess = (new SubprocessInstance($this->projectDir.'/bin/console', 'child', null))->buildProcess()->start();
+        $subprocess->getProcess()->wait();
+
+        $this->assertNull($subprocess->getResult());
+        // Nothing to remove
+        $subprocess->removeBuffer();
+    }
+
+    public function testRemoveBuffer(): void
+    {
+        $subprocess = (new SubprocessInstance($this->projectDir.'/bin/console', 'child', null, [], [SubprocessInstance::OPTION_JSON_BUFFERING => true]))
+            ->buildProcess();
+        $bufferPath = $this->getBufferPath($subprocess);
+        $this->assertFileExists($bufferPath);
+
+        $subprocess->removeBuffer();
+
+        $this->assertFileDoesNotExist($bufferPath);
+    }
+
     public function testMissingConsoleThrows(): void
     {
         $this->expectException(\RuntimeException::class);
@@ -110,4 +153,22 @@ public function testPassingTheKernelIsDeprecated(): void
         // The log directory of the kernel is no longer created
         $this->assertDirectoryDoesNotExist($this->projectDir.'/var');
     }
+
+    private function runWithJsonBuffering(): SubprocessInstance
+    {
+        $subprocess = (new SubprocessInstance($this->projectDir.'/bin/console', 'child', null, [], [SubprocessInstance::OPTION_JSON_BUFFERING => true]))
+            ->buildProcess()
+            ->start();
+        $subprocess->getProcess()->wait();
+        $this->assertSame(0, $subprocess->getProcess()->getExitCode(), $subprocess->getProcess()->getErrorOutput());
+
+        return $subprocess;
+    }
+
+    private function getBufferPath(SubprocessInstance $subprocess): string
+    {
+        preg_match('#--output=([^\']+)#', $subprocess->getProcess()->getCommandLine(), $matches);
+
+        return $matches[1] ?? '';
+    }
 }

tests/Task/Process/ProcessLauncherTaskTest.php

@@ -16,9 +16,11 @@
 use CleverAge\ProcessBundle\Configuration\ProcessConfiguration;
 use CleverAge\ProcessBundle\Configuration\TaskConfiguration;
 use CleverAge\ProcessBundle\Context\ContextualOptionResolver;
+use CleverAge\ProcessBundle\Filesystem\TemporaryFile;
 use CleverAge\ProcessBundle\Model\AbstractConfigurableTask;
 use CleverAge\ProcessBundle\Model\ProcessHistory;
 use CleverAge\ProcessBundle\Model\ProcessState;
+use CleverAge\ProcessBundle\Model\SubprocessInstance;
 use CleverAge\ProcessBundle\Registry\ProcessConfigurationRegistry;
 use CleverAge\ProcessBundle\Task\Process\ProcessLauncherTask;
 use PHPUnit\Framework\TestCase;
@@ -34,8 +36,12 @@
 #[\PHPUnit\Framework\Attributes\UsesClass(ContextualOptionResolver::class)]
 #[\PHPUnit\Framework\Attributes\UsesClass(ProcessHistory::class)]
 #[\PHPUnit\Framework\Attributes\UsesClass(ProcessState::class)]
+#[\PHPUnit\Framework\Attributes\UsesClass(SubprocessInstance::class)]
+#[\PHPUnit\Framework\Attributes\UsesClass(TemporaryFile::class)]
 class ProcessLauncherTaskTest extends TestCase
 {
+    private string $consoleDir;
+
     #[\PHPUnit\Framework\Attributes\DoesNotPerformAssertions]
     public function testDefaultOptionsAreResolved(): void
     {
@@ -87,6 +93,153 @@ public function testPassingTheKernelIsDeprecated(): void
         ]);
     }
 
+    public function testJsonBufferingOutputsThePathOfTheWrittenBuffer(): void
+    {
+        [$task, $state] = $this->launch('write');
+
+        $this->assertTrue($task->next($state));
+        $state->reset(false);
+        $task->execute($state);
+
+        $bufferPath = $state->getOutput();
+        $this->assertIsString($bufferPath);
+        $this->assertStringEqualsFile($bufferPath, "[\"a\"]\n");
+        // Removed by the reader of the path, not by the task
+        unlink($bufferPath);
+    }
+
+    public function testJsonBufferingOutputsNothingWhenTheSubprocessWroteNothing(): void
+    {
+        [$task, $state, $bufferPath] = $this->launch('nothing');
+
+        $this->assertFalse($task->next($state));
+        $state->reset(false);
+        $task->flush($state);
+
+        $this->assertTrue($state->isSkipped());
+        $this->assertNull($state->getOutput());
+        $this->assertFileDoesNotExist($bufferPath);
+    }
+
+    public function testJsonBufferIsRemovedWhenTheSubprocessFails(): void
+    {
+        [$task, $state, $bufferPath] = $this->launch('fail');
+
+        try {
+            $task->next($state);
+            $this->fail('The failure of the sub-process must throw');
+        } catch (\RuntimeException $e) {
+            $this->assertSame('Sub-process has failed: General error', $e->getMessage());
+        }
+        $this->assertFileDoesNotExist($bufferPath);
+    }
+
+    public function testJsonBuffersOfTheStoppedSubprocessesAreRemovedWhenASubprocessFails(): void
+    {
+        [$task, $state, $runningBufferPath] = $this->launch('sleep', wait: false);
+        [, , $failedBufferPath] = $this->launch('fail', $task, $state);
+
+        try {
+            $task->next($state);
+            $this->fail('The failure of the sub-process must throw');
+        } catch (\RuntimeException) {
+        }
+        $this->assertFileDoesNotExist($runningBufferPath);
+        $this->assertFileDoesNotExist($failedBufferPath);
+    }
+
+    public function testJsonBufferOfATerminatedSubprocessIsKeptWhenAnotherSubprocessFails(): void
+    {
+        // The failed sub-process is handled first, the other one is already terminated but not handled yet
+        [$task, $state, $failedBufferPath] = $this->launch('slowfail', wait: false);
+        [, , $bufferPath] = $this->launch('write', $task, $state);
+        $launchedProcesses = (new \ReflectionProperty($task, 'launchedProcesses'))->getValue($task);
+        $this->assertIsArray($launchedProcesses);
+        $failedSubprocess = reset($launchedProcesses);
+        $this->assertInstanceOf(SubprocessInstance::class, $failedSubprocess);
+        $failedSubprocess->getProcess()->wait();
+
+        try {
+            $state->reset(true);
+            $task->execute($state);
+            $this->fail('The failure of the sub-process must throw');
+        } catch (\RuntimeException) {
+        }
+        $this->assertFileDoesNotExist($failedBufferPath);
+        $this->assertFileExists($bufferPath);
+
+        // With the "skip" error strategy, the process goes on: the result of the terminated sub-process is output
+        $this->assertTrue($task->next($state));
+        $state->reset(false);
+        $task->execute($state);
+        $this->assertSame($bufferPath, $state->getOutput());
+        unlink($bufferPath);
+    }
+
+    protected function tearDown(): void
+    {
+        if (isset($this->consoleDir)) {
+            unlink($this->consoleDir.'/console');
+            rmdir($this->consoleDir);
+        }
+    }
+
+    /**
+     * Launches a sub-process with JSON buffering and waits for it. The fake console depends on the input: "write"
+     * writes an array in the buffer, "fail" exits with an error ("slowfail" after 0.5 s), "sleep" keeps running,
+     * otherwise it outputs nothing.
+     *
+     * @return array{ProcessLauncherTask, ProcessState, string} the task, its state and the path of the buffer
+     */
+    private function launch(string $input, ?ProcessLauncherTask $task = null, ?ProcessState $state = null, bool $wait = true): array
+    {
+        if (!isset($this->consoleDir)) {
+            $this->consoleDir = sys_get_temp_dir().'/process_launcher_test_'.uniqid('', true);
+            mkdir($this->consoleDir);
+            file_put_contents($this->consoleDir.'/console', <<<'PHP'
+                <?php
+                $input = stream_get_contents(STDIN);
+                if ('write' === $input) {
+                    file_put_contents(substr($argv[array_key_last($argv) - 2], 9), "[\"a\"]\n");
+                } elseif ('fail' === $input) {
+                    exit(1);
+                } elseif ('slowfail' === $input) {
+                    usleep(500000);
+                    exit(1);
+                } elseif ('sleep' === $input) {
+                    sleep(30);
+                }
+                PHP);
+        }
+        if (!$task instanceof ProcessLauncherTask || !$state instanceof ProcessState) {
+            $task = new ProcessLauncherTask(new NullLogger(), $this->createRegistry(), $this->consoleDir.'/console');
+            $state = $this->createState([
+                'process' => 'child',
+                'json_buffering' => true,
+                'sleep_interval_after_launch' => 0,
+                'sleep_on_finalize_interval' => 0,
+            ]);
+            $task->initialize($state);
+        }
+        $state->reset(true);
+        $state->setInput($input);
+        $task->execute($state);
+        $this->assertTrue($state->isSkipped());
+
+        $launchedProcesses = (new \ReflectionProperty($task, 'launchedProcesses'))->getValue($task);
+        $this->assertIsArray($launchedProcesses);
+        $subprocess = end($launchedProcesses);
+        $this->assertInstanceOf(SubprocessInstance::class, $subprocess);
+        if ($wait) {
+            $subprocess->getProcess()->wait();
+        }
+        preg_match('#--output=([^\']+)#', $subprocess->getProcess()->getCommandLine(), $matches);
+        $bufferPath = $matches[1] ?? '';
+        $this->assertFileExists($bufferPath);
+
+        return [$task, $state, $bufferPath];
+    }
+
     private function createTask(): ProcessLauncherTask
     {
         return new ProcessLauncherTask(new NullLogger(), $this->createRegistry(), '/app/bin/console', 'test');