#331 - ProcessManager: track the processed iterable and blocking tasks by task configuration

CHANGELOG.md

@@ -3,6 +3,7 @@ Latest
 
 ## Fixes
 * [#343](https://github.com/cleverage/process-bundle/issues/343) The temporary files of `FileSplitterTask` (`file_splitter`), `CsvSplitterTask` (`csv_splitter`) and of the JSON buffer of `ProcessLauncherTask` (`json_buffering`) are created with a random name, exclusively (an existing path, symbolic links included, is never reused) and readable by their owner only, instead of a name derived from the time (`uniqid()`), opened in `w` mode with the permissions of the umask. Update documentation, add tests.
+* [#331](https://github.com/cleverage/process-bundle/issues/331) `ProcessManager`: the iterable tasks that produced items and the blocking tasks waiting for `proceed()` are tracked by task configuration instead of task code, and forgotten when their process ends. A task of a sub-process (`process_executor`) with the same code as a task of the parent no longer makes the parent lose the output of its blocking task (without error) or skip the flush at the end of its iteration (merged batches), and a failed sub-process no longer leaves its tasks to its next execution. Add tests.
 
 v5.4.1
 -----

src/Manager/ProcessManager.php

@@ -53,11 +53,15 @@ class ProcessManager implements ProcessManagerInterface, ResetInterface
     protected const EXECUTE_FLUSH = 4;
 
     /**
+     * Indexed by task configuration (spl_object_id()): a sub-process may have a task with the same code.
+     *
      * @var TaskConfiguration[]
      */
     protected array $processedIterables = [];
 
     /**
+     * Indexed by task configuration (spl_object_id()): a sub-process may have a task with the same code.
+     *
      * @var TaskConfiguration[]
      */
     protected array $processedBlockings = [];
@@ -220,6 +224,12 @@ protected function doExecute(string $processCode, mixed $input = null, array $co
             $this->endProcess($processHistory);
 
             throw $error;
+        } finally {
+            // A failed or stopped process may leave tasks in the registries, seen by its next execution
+            foreach ($processConfiguration->getTaskConfigurations() as $taskConfiguration) {
+                $this->removeProcessedIterable($taskConfiguration);
+                $this->removeProcessedBlocking($taskConfiguration);
+            }
         }
 
         $this->processHistory = $parentProcessHistory;
@@ -646,46 +656,46 @@ protected function checkProcess(ProcessConfiguration $processConfiguration): voi
      */
     protected function addProcessedIterable(TaskConfiguration $taskConfiguration): void
     {
-        $this->processedIterables[$taskConfiguration->getCode()] = $taskConfiguration;
+        $this->processedIterables[spl_object_id($taskConfiguration)] = $taskConfiguration;
     }
 
     /**
      * If true this means that the tasks returned an element at least once.
      */
     protected function hasProcessedIterable(TaskConfiguration $taskConfiguration): bool
     {
-        return \array_key_exists($taskConfiguration->getCode(), $this->processedIterables);
+        return \array_key_exists(spl_object_id($taskConfiguration), $this->processedIterables);
     }
 
     /**
      * Once everything was flushed, the task is resolved and can be removed from the stack.
      */
     protected function removeProcessedIterable(TaskConfiguration $taskConfiguration): void
     {
-        unset($this->processedIterables[$taskConfiguration->getCode()]);
+        unset($this->processedIterables[spl_object_id($taskConfiguration)]);
     }
 
     /**
      * Add blocking tasks that were just processed.
      */
     protected function addProcessedBlocking(TaskConfiguration $taskConfiguration): void
     {
-        $this->processedBlockings[$taskConfiguration->getCode()] = $taskConfiguration;
+        $this->processedBlockings[spl_object_id($taskConfiguration)] = $taskConfiguration;
     }
 
     /**
      * If true this means the task was processed normally but was never run with proceed.
      */
     protected function hasProcessedBlocking(TaskConfiguration $taskConfiguration): bool
     {
-        return \array_key_exists($taskConfiguration->getCode(), $this->processedBlockings);
+        return \array_key_exists(spl_object_id($taskConfiguration), $this->processedBlockings);
     }
 
     /**
      * Once a blocking task has been proceeded, we can remove it from the stack.
      */
     protected function removeProcessedBlocking(TaskConfiguration $taskConfiguration): void
     {
-        unset($this->processedBlockings[$taskConfiguration->getCode()]);
+        unset($this->processedBlockings[spl_object_id($taskConfiguration)]);
     }
 }

tests/Manager/ProcessManagerSubProcessTest.php

@@ -0,0 +1,195 @@
+<?php
+
+declare(strict_types=1);
+
+/*
+ * This file is part of the CleverAge/ProcessBundle package.
+ *
+ * Copyright (c) Clever-Age
+ *
+ * For the full copyright and license information, please view the LICENSE
+ * file that was distributed with this source code.
+ */
+
+namespace CleverAge\ProcessBundle\Tests\Manager;
+
+use CleverAge\ProcessBundle\Configuration\ProcessConfiguration;
+use CleverAge\ProcessBundle\Configuration\TaskConfiguration;
+use CleverAge\ProcessBundle\Context\ContextualOptionResolver;
+use CleverAge\ProcessBundle\Event\ProcessEvent;
+use CleverAge\ProcessBundle\Logger\AbstractLogger;
+use CleverAge\ProcessBundle\Logger\ProcessLogger;
+use CleverAge\ProcessBundle\Logger\TaskLogger;
+use CleverAge\ProcessBundle\Manager\ProcessManager;
+use CleverAge\ProcessBundle\Model\AbstractConfigurableTask;
+use CleverAge\ProcessBundle\Model\ProcessHistory;
+use CleverAge\ProcessBundle\Model\ProcessState;
+use CleverAge\ProcessBundle\Model\TaskInterface;
+use CleverAge\ProcessBundle\Registry\ProcessConfigurationRegistry;
+use CleverAge\ProcessBundle\Registry\TaskRegistry;
+use CleverAge\ProcessBundle\Task\AbstractIterableOutputTask;
+use CleverAge\ProcessBundle\Task\AggregateIterableTask;
+use CleverAge\ProcessBundle\Task\ConstantIterableOutputTask;
+use CleverAge\ProcessBundle\Task\SimpleBatchTask;
+use PHPUnit\Framework\TestCase;
+use Psr\Log\NullLogger;
+use Symfony\Component\DependencyInjection\Container;
+use Symfony\Component\EventDispatcher\EventDispatcher;
+
+/**
+ * A sub-process sharing the process manager of its parent (as the process_executor task) must not change the state of
+ * the tasks of the parent, even when its tasks have the same codes.
+ */
+#[\PHPUnit\Framework\Attributes\CoversClass(ProcessManager::class)]
+#[\PHPUnit\Framework\Attributes\UsesClass(ProcessConfiguration::class)]
+#[\PHPUnit\Framework\Attributes\UsesClass(TaskRegistry::class)]
+#[\PHPUnit\Framework\Attributes\UsesClass(TaskConfiguration::class)]
+#[\PHPUnit\Framework\Attributes\UsesClass(ContextualOptionResolver::class)]
+#[\PHPUnit\Framework\Attributes\UsesClass(ProcessEvent::class)]
+#[\PHPUnit\Framework\Attributes\UsesClass(AbstractLogger::class)]
+#[\PHPUnit\Framework\Attributes\UsesClass(AbstractConfigurableTask::class)]
+#[\PHPUnit\Framework\Attributes\UsesClass(ProcessHistory::class)]
+#[\PHPUnit\Framework\Attributes\UsesClass(ProcessState::class)]
+#[\PHPUnit\Framework\Attributes\UsesClass(ProcessConfigurationRegistry::class)]
+#[\PHPUnit\Framework\Attributes\UsesClass(AbstractIterableOutputTask::class)]
+#[\PHPUnit\Framework\Attributes\UsesClass(ConstantIterableOutputTask::class)]
+#[\PHPUnit\Framework\Attributes\UsesClass(AggregateIterableTask::class)]
+#[\PHPUnit\Framework\Attributes\UsesClass(SimpleBatchTask::class)]
+class ProcessManagerSubProcessTest extends TestCase
+{
+    private ProcessManager $processManager;
+
+    /**
+     * @var list<mixed>
+     */
+    private array $recorded = [];
+
+    protected function setUp(): void
+    {
+        $container = new Container();
+        $container->set('test.parent_iterate', new ConstantIterableOutputTask());
+        $container->set('test.child_iterate', new ConstantIterableOutputTask());
+        $container->set('test.parent_aggregate', new AggregateIterableTask());
+        $container->set('test.child_aggregate', new AggregateIterableTask());
+        $container->set('test.batch', new SimpleBatchTask());
+        // Runs the given sub-process and forwards its input, as the process_executor task
+        $container->set('test.sub_process', new class($this) implements TaskInterface {
+            public function __construct(
+                private readonly ProcessManagerSubProcessTest $test,
+            ) {
+            }
+
+            public function execute(ProcessState $state): void
+            {
+                $this->test->executeSubProcess((string) $state->getTaskConfiguration()->getOption('process'));
+                $state->setOutput($state->getInput());
+            }
+        });
+        $container->set('test.record', new class($this) implements TaskInterface {
+            public function __construct(
+                private readonly ProcessManagerSubProcessTest $test,
+            ) {
+            }
+
+            public function execute(ProcessState $state): void
+            {
+                $this->test->record($state->getInput());
+            }
+        });
+
+        $registry = new ProcessConfigurationRegistry(
+            [
+                'test.blocking_parent' => $this->createProcessConfiguration([
+                    'iterate' => $this->createTaskConfiguration('@test.parent_iterate', ['output' => [1, 2, 3]], ['sub_process']),
+                    'sub_process' => $this->createTaskConfiguration('@test.sub_process', ['process' => 'test.blocking_child'], ['aggregate']),
+                    'aggregate' => $this->createTaskConfiguration('@test.parent_aggregate'),
+                ], 'aggregate'),
+                'test.blocking_child' => $this->createProcessConfiguration([
+                    'iterate' => $this->createTaskConfiguration('@test.child_iterate', ['output' => ['a']], ['aggregate']),
+                    'aggregate' => $this->createTaskConfiguration('@test.child_aggregate'),
+                ]),
+                'test.iterable_parent' => $this->createProcessConfiguration([
+                    'iterate' => $this->createTaskConfiguration('@test.parent_iterate', ['output' => [1, 2, 3, 4, 5]], ['sub_process']),
+                    'sub_process' => $this->createTaskConfiguration('@test.sub_process', ['process' => 'test.iterable_child'], ['batch']),
+                    'batch' => $this->createTaskConfiguration('@test.batch', ['batch_count' => 2], ['record']),
+                    'record' => $this->createTaskConfiguration('@test.record'),
+                ]),
+                'test.iterable_child' => $this->createProcessConfiguration([
+                    'iterate' => $this->createTaskConfiguration('@test.child_iterate', ['output' => ['a']]),
+                ]),
+            ],
+            'stop'
+        );
+
+        $this->processManager = new ProcessManager(
+            $container,
+            new ProcessLogger(new NullLogger()),
+            new TaskLogger(new NullLogger()),
+            $registry,
+            new ContextualOptionResolver(),
+            new EventDispatcher(),
+            taskRegistry: new TaskRegistry($container),
+        );
+    }
+
+    public function executeSubProcess(string $processCode): void
+    {
+        $this->processManager->execute($processCode);
+    }
+
+    public function record(mixed $value): void
+    {
+        $this->recorded[] = $value;
+    }
+
+    public function testABlockingTaskOfTheParentIsProceededAfterASubProcessWithATaskOfTheSameCode(): void
+    {
+        self::assertSame([1, 2, 3], $this->processManager->execute('test.blocking_parent'));
+    }
+
+    public function testAnIterableTaskOfTheParentIsFlushedAfterASubProcessWithATaskOfTheSameCode(): void
+    {
+        $this->processManager->execute('test.iterable_parent');
+
+        self::assertSame([[1, 2], [3, 4], [5]], $this->recorded);
+    }
+
+    /**
+     * @param array<string, array<string, mixed>> $tasks
+     *
+     * @return array<string, mixed>
+     */
+    private function createProcessConfiguration(array $tasks, ?string $endPoint = null): array
+    {
+        return [
+            'options' => [],
+            'entry_point' => null,
+            'end_point' => $endPoint,
+            'description' => '',
+            'help' => '',
+            'public' => true,
+            'tasks' => $tasks,
+        ];
+    }
+
+    /**
+     * @param array<string, mixed> $options
+     * @param list<string>         $outputs
+     *
+     * @return array<string, mixed>
+     */
+    private function createTaskConfiguration(string $service, array $options = [], array $outputs = []): array
+    {
+        return [
+            'service' => $service,
+            'options' => $options,
+            'description' => '',
+            'help' => '',
+            'outputs' => $outputs,
+            'errors' => [],
+            'error_outputs' => [],
+            'error_strategy' => null,
+            'log_level' => null,
+        ];
+    }
+}