#307 - InputAggregatorTask: drop the items skipped by a branch, remove the deprecation

CHANGELOG.md

@@ -2,7 +2,7 @@ Latest
 ------
 
 ## Changes
-* [#305](https://github.com/cleverage/process-bundle/issues/305) Bundle best practices, residues: the generic transformers are no longer autowired (explicit arguments). `ProcessHelpCommand` receives the `TaskRegistry` instead of the service container (passing a container is deprecated). Fetching from the service container a task of the application made public by the `TaskInterface` autoconfiguration is deprecated (`container.private` tag, new internal `PublicTaskDeprecationCompilerPass`); the tasks made public by the application itself are not concerned. Trigger the missing deprecation notices: `InputAggregatorTask`, `CheckSerializerCompilerPass`, `errors` task configuration key (use `error_outputs`). `TaskConfiguration::getErrors()` triggers a single notice on PHP 8.4+. `CleverAgeProcessBundle` is marked `@final since 5.4`. Update documentation, add tests.
+* [#305](https://github.com/cleverage/process-bundle/issues/305) Bundle best practices, residues: the generic transformers are no longer autowired (explicit arguments). `ProcessHelpCommand` receives the `TaskRegistry` instead of the service container (passing a container is deprecated). Fetching from the service container a task of the application made public by the `TaskInterface` autoconfiguration is deprecated (`container.private` tag, new internal `PublicTaskDeprecationCompilerPass`); the tasks made public by the application itself are not concerned. Trigger the missing deprecation notices: `CheckSerializerCompilerPass`, `errors` task configuration key (use `error_outputs`). `TaskConfiguration::getErrors()` triggers a single notice on PHP 8.4+. `CleverAgeProcessBundle` is marked `@final since 5.4`. Update documentation, add tests.
 * [#299](https://github.com/cleverage/process-bundle/issues/299) Require `symfony/polyfill-php83`, `symfony/polyfill-php84` and `symfony/polyfill-php85`: the functions of PHP 8.3 to 8.5 can be used while supporting PHP 8.2. Rector targets PHP 8.2 (`withPhpVersion()`, `withPhpSets()` read from `composer.json`) and applies the rules using a newer function when its polyfill is required: `ProcessConfiguration` uses `array_any()` again; `array_first()` replaces `reset()` / `current()` in `ProcessConfiguration::getMainTask()`, `ConstraintLoader` and `ProcessHelpCommand` (no internal array pointer, same results).
 * [#279](https://github.com/cleverage/process-bundle/issues/279) Define the public API (Symfony BC promise): the classes not meant to be extended are marked `@final since 5.4` (the engine: `ProcessManager`, `ProcessConfigurationRegistry`, `TransformerRegistry`, `ContextualOptionResolver`; the models: `ProcessState`, `ProcessHistory`, `SubprocessInstance`, the configurations, the events, the `Filesystem` files; the concrete tasks and transformers), they will be `final` in v6.0: extending them triggers a deprecation with the Symfony `DebugClassLoader`. The abstract classes, interfaces, traits, exceptions and the file tasks extended by other tasks (`FileReaderTask`, `LineReaderTask`, `FolderBrowserTask`, `CsvReaderTask`, `InputCsvReaderTask`, `Filesystem\CsvResource`) stay open. `TaskRegistry`, new in this version, is `final`. The infrastructure classes are marked `@internal`: `CleverAgeProcessExtension`, `Configuration`, the commands, `DataQueueEventListener`, `PhpFunctionProvider`, the Monolog processors, `ProcessLogger` and `TaskLogger`. Update documentation.
 * [#269](https://github.com/cleverage/process-bundle/issues/269) Composer: `symfony/process`, `symfony/stopwatch`, `symfony/var-dumper` and `symfony/monolog-bundle` are no longer required, they are suggested (applications using them must require them). Without `symfony/process` (`CommandRunnerTask`, `ProcessLauncherTask`) or `symfony/stopwatch` (`StopwatchTask`), the container still compiles and these tasks fail with an explicit message (`UnavailableServicesCompilerPass`); without `symfony/var-dumper`, nothing is dumped (as before); without `symfony/monolog-bundle`, the `cleverage_process*` channels and processors are not registered and the default logger is used. `monolog/monolog` stays required (`Logger\*Processor`). Remove the unused `symfony/monolog-bridge` dependency. `ProcessConfiguration` no longer uses `array_any()` (PHP 8.4), whose polyfill was only installed by the removed packages. Add a `conflict` with `cleverage/doctrine-process-bundle`, `cleverage/soap-process-bundle` and `cleverage/ui-process-bundle` `<3.3`, which use the MonologBundle without requiring it. Update documentation, add tests.
@@ -20,6 +20,7 @@ Latest
 * [#268](https://github.com/cleverage/process-bundle/issues/268) Clean up composer.json: remove the unused `symfony/dotenv`, `symfony/runtime` and `symfony/error-handler` dependencies (applications relying on the bundle to install them must require them), bound `symfony/service-contracts` to `^2.5|^3`, update the package description.
 
 ## Fixes
+* [#307](https://github.com/cleverage/process-bundle/issues/307) Fix InputAggregatorTask: when a branch skips an item (filter, error with the `skip` strategy), the inputs received from the other branches for this item are dropped instead of being joined with the inputs of the next item. The inputs of an item are told apart by the executions of their common ancestors (new `ProcessState::getExecutionCount()`, incremented by the `ProcessManager` before each execution of a task). InputAggregatorTask is no longer deprecated (its limits described in 2018 no longer exist) and is marked `@final since 5.4`. Update documentation, add tests.
 * [#301](https://github.com/cleverage/process-bundle/issues/301) Fix ProcessManager: the history of the parent process is restored after a failed process too, so that the logs of a parent process going on after a failed sub-process (e.g. `process_executor` task with the `skip` error strategy) are no longer attributed to the sub-process (`process_id`, `process_code` and `process_context` added by the Monolog processors), and that no history of a failed process is left after its execution (`getProcessHistory()`). The `Critical process failure` log and the `cleverage_process.fail` event are still attributed to the failed process. `ProcessManager` implements `ResetInterface` (`kernel.reset` tag): its state is reset between two messages of a long-running worker. Add tests.
 
 v5.3

docs/05-good_practices.md

@@ -71,7 +71,8 @@ See [errors and skips](04-advanced_workflow.md#errors-and-skips).
 
 When a task has several `outputs`, each output is fully processed one after the other, in the configured order, but
 this is an implementation detail. If a task needs the result of another branch, make the dependency explicit in the
-graph (e.g. with a blocking task) rather than relying on the order in which branches are executed.
+graph (e.g. with a blocking task or an [InputAggregatorTask](reference/tasks/input_aggregator_task.md)) rather than
+relying on the order in which branches are executed.
 
 ## Use the context for runtime parameters
 

docs/reference/tasks/input_aggregator_task.md

@@ -1,13 +1,13 @@
 InputAggregatorTask
 ===================
 
-**Deprecated since v5.4, removed in v6.0**: this task is too error-prone (it should be refactored as a blocking task).
-An `E_USER_DEPRECATED` error is triggered when it is instantiated.
-
-Accumulates inputs coming from several parent tasks. The task is skipped until an input has been received from every
+Joins the inputs coming from several parent tasks. The task is skipped until an input has been received from every
 parent declared in `input_codes`; it then outputs an array of all received inputs, indexed by their destination key,
 and clears its buffer (except for the keys listed in `keep_inputs`).
 
+Unlike a blocking task (e.g. [AggregateIterableTask](aggregate_iterable_task.md)), which waits for all the items and
+outputs them once, it outputs once **per item** in an iterable process.
+
 Task reference
 --------------
 
@@ -62,10 +62,42 @@ aggregate:
       branch_b: b
 ```
 
+* Join two branches for each item of an iterable process, a branch skipping some items
+
+```yaml
+# Task configuration level
+entry:
+  task: constant_iterable_output
+  options:
+    output: [1, 2, 3]
+  outputs: [branch_a, branch_b]
+branch_a:
+  task: filter
+  options:
+    not_match: { '': 2 }
+  outputs: [aggregate]
+branch_b:
+  task: transformer
+  options:
+    transformers:
+      sprintf: { format: 'item %s' }
+  outputs: [aggregate]
+aggregate:
+  task: input_aggregator
+  options:
+    input_codes:
+      branch_a: a
+      branch_b: b
+# Outputs ['a' => 1, 'b' => 'item 1'], then ['a' => 3, 'b' => 'item 3']: the item 2, skipped by branch_a, is dropped
+```
+
 Notes
 -----
 
-In iterable processes, inputs of different iterations may get mixed (for instance when a parent task skips an item
-because of an error), even with `clean_input_on_override: true`. Its usage is **strongly discouraged** outside of
-non-iterable processes; prefer a blocking task such as [ArrayMergeTask](array_merge_task.md) or
-[AggregateIterableTask](aggregate_iterable_task.md) when possible.
+The inputs of an item are told apart by the executions of the ancestors they have in common (their execution count,
+`ProcessState::getExecutionCount()`): when an input arrives after a common ancestor has been executed again, the inputs
+received from the other branches for a previous item are dropped (except those of `keep_inputs`). An item skipped by a
+branch (filter, error with `error_strategy: skip`) is therefore never joined with the next item. Before v5.4, the
+inputs of two items could be joined in that case.
+
+Inputs without any common ancestor (e.g. two branches starting with their own root task) are always joined.

src/Manager/ProcessManager.php

@@ -413,6 +413,7 @@ protected function processExecution(TaskConfiguration $taskConfiguration, int $e
             return;
         }
 
+        $state->incrementExecutionCount();
         try {
             if (self::EXECUTE_PROCESS === $executionFlag) {
                 $state->reset(false);

src/Model/ProcessState.php

@@ -78,6 +78,11 @@ class ProcessState
 
     protected string $status = self::STATUS_NEW;
 
+    /**
+     * Number of executions of the task (execute, proceed or flush), not reset between the items: tells the items apart.
+     */
+    protected int $executionCount = 0;
+
     public function __construct(
         protected ProcessConfiguration $processConfiguration,
         protected ProcessHistory $processHistory,
@@ -258,6 +263,23 @@ public function setPreviousState(?self $previousState): void
         $this->previousState = $previousState;
     }
 
+    /**
+     * Number of executions of the task (execute, proceed or flush) since the beginning of the process: each output of a
+     * task follows an execution, so a task downstream can tell whether two inputs come from the same item.
+     */
+    public function getExecutionCount(): int
+    {
+        return $this->executionCount;
+    }
+
+    /**
+     * @internal called by the ProcessManager before each execution of the task
+     */
+    public function incrementExecutionCount(): void
+    {
+        ++$this->executionCount;
+    }
+
     public function getStatus(): string
     {
         return $this->status;

src/Task/InputAggregatorTask.php

@@ -18,24 +18,29 @@
 use Symfony\Component\OptionsResolver\OptionsResolver;
 
 /**
- * Wait for defined inputs before passing an aggregated output.
- * Should have been a BlockingTask, but due to limitations in the current model, it's a hack using skips and finalize.
+ * Joins the outputs of several branches: waits for an input from every parent task declared in input_codes, then
+ * outputs them indexed by their destination key. Unlike a blocking task, it outputs once per item.
  *
- * @see        README.md:Known issues
- * @deprecated since 5.4, it's way too much error prone - should be refactored as a blocking task, will be removed in v6.0
+ * The inputs of an item are told apart by the executions of their common ancestors: when a branch skips an item, the
+ * inputs already received for this item are dropped when the inputs of the next item arrive, instead of being joined
+ * with them.
+ *
+ * @final since 5.4
  */
 class InputAggregatorTask extends AbstractConfigurableTask
 {
-    public function __construct()
-    {
-        trigger_deprecation('cleverage/process-bundle', '5.4', 'The "%s" task is deprecated: it is too error-prone. It will be removed in v6.0.', self::class);
-    }
-
     /**
      * @var array<string, mixed>
      */
     protected array $inputs = [];
 
+    /**
+     * Execution count of the ancestors of each received input (ancestor state id => execution count), by input code.
+     *
+     * @var array<string, array<int, int>>
+     */
+    protected array $inputAncestors = [];
+
     /**
      * Store inputs and once everything has been received, pass to next task
      * Once an output has been generated this task is reset, and may wait for another loop.
@@ -48,15 +53,19 @@ public function execute(ProcessState $state): void
         }
 
         $inputCode = $this->getInputCode($state);
+        $ancestors = $this->getAncestorExecutions($previousState);
+        $this->removeInputsOfPreviousItems($state, $inputCode, $ancestors);
         if (\array_key_exists($inputCode, $this->inputs)) {
             if ($this->getOption($state, 'clean_input_on_override')) {
                 $this->inputs = [];
+                $this->inputAncestors = [];
             } else {
                 throw new \UnexpectedValueException("The output from input '{$inputCode}' has already been defined, please use an aggregator if you have an iterable output");
             }
         }
 
         $this->inputs[$inputCode] = $state->getInput();
+        $this->inputAncestors[$inputCode] = $ancestors;
 
         if ($this->isResolved($state)) {
             $state->setOutput($this->inputs);
@@ -66,7 +75,7 @@ public function execute(ProcessState $state): void
                 if (null !== $keepInputs && \in_array($inputCode, $keepInputs, true)) {
                     continue;
                 }
-                unset($this->inputs[$inputCode]);
+                unset($this->inputs[$inputCode], $this->inputAncestors[$inputCode]);
             }
         } else {
             $state->setSkipped(true);
@@ -118,4 +127,45 @@ protected function isResolved(ProcessState $state): bool
 
         return true;
     }
+
+    /**
+     * Execution count of the input's ancestors, from its parent task up to the first task of its branch.
+     *
+     * @return array<int, int> ancestor state id => execution count
+     */
+    protected function getAncestorExecutions(ProcessState $previousState): array
+    {
+        $ancestors = [];
+        for ($ancestor = $previousState; $ancestor instanceof ProcessState; $ancestor = $ancestor->getPreviousState()) {
+            $id = spl_object_id($ancestor);
+            if (\array_key_exists($id, $ancestors)) {
+                break;
+            }
+            $ancestors[$id] = $ancestor->getExecutionCount();
+        }
+
+        return $ancestors;
+    }
+
+    /**
+     * Removes the inputs of the other branches received for a previous item: a common ancestor has been executed again
+     * since they were received (e.g. a branch skipped this item). The inputs of keep_inputs are kept.
+     *
+     * @param array<int, int> $ancestors
+     */
+    protected function removeInputsOfPreviousItems(ProcessState $state, string $inputCode, array $ancestors): void
+    {
+        $keepInputs = $this->getOption($state, 'keep_inputs') ?? [];
+        foreach ($this->inputAncestors as $receivedInputCode => $receivedAncestors) {
+            if ($receivedInputCode === $inputCode || \in_array($receivedInputCode, $keepInputs, true)) {
+                continue;
+            }
+            foreach (array_intersect_key($receivedAncestors, $ancestors) as $id => $executionCount) {
+                if ($executionCount !== $ancestors[$id]) {
+                    unset($this->inputs[$receivedInputCode], $this->inputAncestors[$receivedInputCode]);
+                    break;
+                }
+            }
+        }
+    }
 }

tests/App/TestKernel.php

@@ -111,6 +111,51 @@ protected function configureContainer(ContainerConfigurator $container): void
                         ],
                     ],
                 ],
+                // branch_a skips the item 2, branch_b the item 4
+                'test.input_aggregator' => [
+                    'end_point' => 'collect',
+                    'tasks' => [
+                        'entry' => [
+                            'task' => 'constant_iterable_output',
+                            'options' => ['output' => [1, 2, 3, 4, 5]],
+                            'outputs' => ['branch_a', 'branch_b'],
+                        ],
+                        'branch_a' => ['task' => 'filter', 'options' => ['not_match' => ['' => 2]], 'outputs' => ['format_a']],
+                        'format_a' => ['task' => 'transformer', 'options' => ['transformers' => ['sprintf' => ['format' => 'a%s']]], 'outputs' => ['aggregate']],
+                        'branch_b' => ['task' => 'filter', 'options' => ['not_match' => ['' => 4]], 'outputs' => ['format_b']],
+                        'format_b' => ['task' => 'transformer', 'options' => ['transformers' => ['sprintf' => ['format' => 'b%s']]], 'outputs' => ['aggregate']],
+                        'aggregate' => [
+                            'task' => 'input_aggregator',
+                            'options' => ['input_codes' => ['format_a' => 'a', 'format_b' => 'b']],
+                            'outputs' => ['collect'],
+                        ],
+                        'collect' => ['task' => 'aggregate_iterable'],
+                    ],
+                ],
+                // branch_a fails on the item "x" (intdiv() of a string), and skips it
+                'test.input_aggregator_error' => [
+                    'end_point' => 'collect',
+                    'tasks' => [
+                        'entry' => [
+                            'task' => 'constant_iterable_output',
+                            'options' => ['output' => [1, 'x', 3]],
+                            'outputs' => ['branch_a', 'branch_b'],
+                        ],
+                        'branch_a' => [
+                            'task' => 'transformer',
+                            'options' => ['transformers' => ['callback' => ['callback' => 'intdiv', 'right_parameters' => [1]]]],
+                            'error_strategy' => 'skip',
+                            'outputs' => ['aggregate'],
+                        ],
+                        'branch_b' => ['task' => 'transformer', 'options' => ['transformers' => ['sprintf' => ['format' => 'b%s']]], 'outputs' => ['aggregate']],
+                        'aggregate' => [
+                            'task' => 'input_aggregator',
+                            'options' => ['input_codes' => ['branch_a' => 'a', 'branch_b' => 'b']],
+                            'outputs' => ['collect'],
+                        ],
+                        'collect' => ['task' => 'aggregate_iterable'],
+                    ],
+                ],
             ],
         ]);