InputAggregatorTask: remove the deprecation, fix the items mixed when a branch skips one #307

Description

InputAggregatorTask (input_aggregator) has been @deprecated since 2018 (236011a, v2.0.0: "It's way too much error prone - should be refactored as a blocking task", @see README.md:Known issues). #305 added a runtime notice (5.4, not released yet) and #292 plans its removal in v6.0. This task is widely used, and it cannot be removed: remove the deprecation, and fix the task instead.

The reasons given in 2018 no longer apply

The "Known issues" section of the README (removed since then: the @see points to nothing) described limits of the blocking tasks in multi-branch workflows. The ProcessManager now proceeds a blocking task in resolve(), once all its parents are resolved in the graph. Checked on main @ a32602b with demo processes (cleverage/process-bundle-demo, aggregate_iterable):

Also, a blocking task is not a replacement: InputAggregatorTask joins, for each item, the outputs of several branches into one array keyed by branch ({"a": …, "b": …}), while a blocking task (aggregate_iterable, array_merge…) waits for all the items and outputs them once. The reference page (Notes) wrongly suggests ArrayMergeTask / AggregateIterableTask instead.

The actual defect: items mixed when a branch skips one

demo.input_aggregator.iterable (items 1 to 3, two branches): {"a":"a1","b":"b1"}, {"a":"a2","b":"b2"}, {"a":"a3","b":"b3"} → OK.

demo.input_aggregator.skipped_branch (items 1 to 5, branch_a skips the item 2, branch_b skips the item 4):

{"a":"a1","b":"b1"}
{"b":"b2","a":"a3"}   ← b of item 2 joined with a of item 3
{"b":"b3","a":"a4"}   ← shifted until the next skip
{"a":"a5","b":"b5"}

Expected: {"a":"a1","b":"b1"}, {"a":"a3","b":"b3"}, {"a":"a5","b":"b5"} (an incomplete item is dropped). The buffer only tells the inputs apart by branch, not by item: clean_input_on_override only helps when the same branch sends twice. The same applies to an error in a branch with error_strategy: skip (read in the code, not executed: ProcessManager::processExecution() sets the same skipped state as FilterTask).

Proposal

1. Remove the deprecation (before the 5.4 release, so that no release ships the notice of Bundle best practices: residues (generic transformers autowiring, help command container, deprecation notices) #305): constructor and trigger_deprecation(), @deprecated, dangling @see README.md:Known issues; reference page (Deprecated mention, Notes), good practices page (mention it again for joining branches), CHANGELOG entry of Bundle best practices: residues (generic transformers autowiring, help command container, deprecation notices) #305, InputAggregatorTaskTest::testTaskIsDeprecated(), KernelTest; remove the item from v6.0: remove the deprecated code #292.
2. Fix the mixing: let the task know which item an input belongs to, and clear the buffer when an input of another item arrives.
   • ProcessState: new execution counter, incremented by ProcessManager::processExecution() on each execute() (getExecutionCount(), @internal or public).
   • InputAggregatorTask: identifies the item by the nearest iterable ancestor of the input (getPreviousState() chain up to an IterableTaskInterface task) and its counter. An input of another item clears the buffer, so an incomplete item is dropped instead of being joined with the next one.
   • Limit to document: branches containing their own iterable task (fan-out inside a branch) have different nearest iterable ancestors → keep the current behaviour (no common ancestor found).
   • Alternative without a core change: detect a new item when an input comes from a branch placed before the last received one in input_codes (outputs are processed depth-first, in order). Simpler, but it depends on the order of input_codes matching the order of the outputs.
3. Tests: the mixing case (skip in the first branch, skip in the last branch, error with skip), non-iterable process unchanged, keep_inputs.

Requirements

• Documentation updates
  • Reference
  • Cookbooks
  • Changelog
• Unit tests

Breaking changes

None. Removing a deprecation that was never released. The fix only changes outputs that are wrong today (mixed items): they are dropped instead.