#481 - A stop only stops the flow it happened in

CHANGELOG.md

@@ -8,6 +8,7 @@ Latest
 * [#483](https://github.com/cleverage/process-bundle/issues/483) Documentation: an exception thrown by `IterableTaskInterface::next()` makes the process fail right away with this exception, the `error_strategy` and the `error_outputs` of the task do not apply.
 
 ## Fixes
+* [#481](https://github.com/cleverage/process-bundle/issues/481) A stop without error (`StopTask`, `ProcessState::stop()`) only stops the flow it happened in, as documented: the outputs of another root task, of a blocking task proceeded or of a flushable task flushed after the stop go on through the tasks of the stopped flow, which stayed stopped and dropped them silently. An iterable task whose iteration has been interrupted by the stop, and a task whose initialization has failed (deprecated), stay stopped. Update documentation, add tests.
 * [#475](https://github.com/cleverage/process-bundle/issues/475) Generic transformers: restore, deprecated, the former names (with underscores instead of dashes) of their transformer codes and contextual options, which no longer worked since 5.9 ([#438](https://github.com/cleverage/process-bundle/issues/438)): a transformer code with a dash that is only registered with underscores (`array-first` for `array_first`), and a contextual option with a dash given or used by its former name (`the_prefix`, `{{ the_prefix }}` for `the-prefix`). Update documentation, add tests.
 * [#491](https://github.com/cleverage/process-bundle/issues/491) `CounterTask`: `flush_every: 0` is rejected when the options are resolved, with a clear message, instead of failing on the first execution with `Modulo by zero`; a negative `flush_every`, which works as its absolute value, is deprecated. Update documentation, add tests.
 * [#477](https://github.com/cleverage/process-bundle/issues/477) "Stop the process on the first invalid item" (`error_strategy: skip` and an error branch ending on a `StopTask`) ends as stopped when the failing task is finalizable (`csv_reader`, `xml_stream_reader`, `stat_counter`...): the error of the skipped item, kept by its state, was taken for a finalization failure and the process ended as failed. Add tests.

UPGRADE.md

@@ -28,6 +28,13 @@ cleverage/process-bundle 5.10: The transformer "array-first" of the generic tran
 "the-prefix" of the generic transformer...`), and will not work in 6.0: write the code of the transformer as registered
 (`array_first`), and the contextual option as defined in `contextual_options` (`the-prefix`, `{{ the-prefix }}`).
 
+### A stop only stops its flow
+
+A stop without error (a `StopTask`, or a task calling `ProcessState::stop()`) only stops the flow it happened in, as
+documented. The outputs of another root task, of a blocking task proceeded or of a flushable task flushed after the stop
+now go on through the tasks of the stopped flow: they were dropped silently by the first task that had stopped. Check
+the processes with several root tasks, or with a blocking task (e.g. `aggregate_iterable`) fed before a stop.
+
 ## v5.8
 
 ### Unknown task references

docs/04-advanced_workflow.md

@@ -162,7 +162,9 @@ A task may also control the flow without any error:
 - `ProcessState::setStopped(true)` / `ProcessState::stop()`: without exception, only the current flow stops: the stop
   signal goes back up to the root task, which stops iterating. The rest of the resolution still happens: other root
   tasks are executed, blocking tasks that already received inputs are proceeded, flushable tasks are flushed and every
-  task is finalized. It is not a failure for the process manager: the `cleverage_process.end` event is dispatched (not
+  task is finalized. Their outputs go on through the tasks of the stopped flow, except through an iterable task whose
+  iteration has been interrupted by the stop: it stays stopped (it would resume its interrupted iteration), and stops
+  the flows reaching it. It is not a failure for the process manager: the `cleverage_process.end` event is dispatched (not
   `cleverage_process.fail`) and the command exits with code 0. The [StopTask](reference/tasks/stop_task.md) also flags
   the process history as stopped: the process ends as stopped on purpose, not as a failure
 

src/Manager/ProcessManager.php

@@ -108,6 +108,14 @@ class ProcessManager implements ProcessManagerInterface, ResetInterface
      */
     protected ?\WeakMap $loggedStates = null;
 
+    /**
+     * Tasks whose initialization has failed (deprecated since 5.1), indexed by task configuration (spl_object_id()):
+     * they stay stopped.
+     *
+     * @var array<int, TaskConfiguration>
+     */
+    protected array $failedInitializations = [];
+
     protected ?ProcessHistory $processHistory = null;
 
     protected ?TaskConfiguration $taskConfiguration = null;
@@ -157,6 +165,7 @@ public function reset(): void
         $this->locks = [];
         $this->endPointOutputs = new \WeakMap();
         $this->loggedStates = null;
+        $this->failedInitializations = [];
         $this->processHistory = null;
         $this->taskConfiguration = null;
     }
@@ -353,6 +362,7 @@ protected function doExecute(string $processCode, mixed $input = null, array $co
             foreach ($processConfiguration->getTaskConfigurations() as $taskConfiguration) {
                 $this->removeProcessedIterable($taskConfiguration);
                 $this->removeProcessedBlocking($taskConfiguration);
+                unset($this->failedInitializations[spl_object_id($taskConfiguration)]);
             }
         }
 
@@ -427,6 +437,7 @@ protected function initialize(TaskConfiguration $taskConfiguration): void
                 ];
                 $this->taskLogger->critical($e->getMessage(), $logContext);
                 $state->stop($e);
+                $this->failedInitializations[spl_object_id($taskConfiguration)] = $taskConfiguration;
                 // @deprecated since 5.1, in v6.0 the process will fail right away, before any task is executed
                 trigger_deprecation(
                     'cleverage/process-bundle',
@@ -551,6 +562,17 @@ protected function processExecution(TaskConfiguration $taskConfiguration, int $e
         }
         $state = $taskConfiguration->getState();
 
+        // A stop only stops the flow it happened in: a later execution of the task (from another root task, by a
+        // blocking task proceeded or a flushable task flushed after the stop...) goes on. Except for a task whose
+        // initialization has failed (deprecated), and for an iterable task, whose iteration has been interrupted (it
+        // would resume it)
+        if ($state->isStopped()
+            && !$task instanceof IterableTaskInterface
+            && !isset($this->failedInitializations[spl_object_id($taskConfiguration)])
+        ) {
+            $state->setStopped(false);
+        }
+
         $this->refreshLocks();
         if ($this->isStopRequested($state)) {
             $state->reset(self::EXECUTE_PROCESS !== $executionFlag);

tests/Manager/ProcessManagerFlowTest.php

@@ -25,6 +25,7 @@
 use CleverAge\ProcessBundle\Manager\ProcessManager;
 use CleverAge\ProcessBundle\Model\BlockingTaskInterface;
 use CleverAge\ProcessBundle\Model\FlushableTaskInterface;
+use CleverAge\ProcessBundle\Model\InitializableTaskInterface;
 use CleverAge\ProcessBundle\Model\IterableTaskInterface;
 use CleverAge\ProcessBundle\Model\ProcessHistory;
 use CleverAge\ProcessBundle\Model\ProcessState;
@@ -171,6 +172,80 @@ public function testStopInAnErrorBranchIsPropagated(): void
         self::assertSame(['check:execute:1', 'check:execute:2', 'errors:execute:2'], $this->getExecutions());
     }
 
+    /**
+     * A stop only stops the flow it happened in: another root task goes on through the tasks of the stopped flow (its
+     * items were dropped by the task that had stopped).
+     */
+    public function testAnotherRootTaskGoesOnThroughTheTasksOfAStoppedFlow(): void
+    {
+        $this->execute([
+            'first' => ['type' => 'iterable', 'items' => [1, 2], 'outputs' => ['shared']],
+            'second' => ['type' => 'iterable', 'items' => [3, 4], 'outputs' => ['shared']],
+            'shared' => ['outputs' => ['check']],
+            'check' => ['stop_on' => 1, 'outputs' => ['after']],
+            'after' => [],
+        ]);
+
+        self::assertSame(
+            ['shared:execute:1', 'check:execute:1', 'shared:execute:3', 'check:execute:3', 'after:execute:3', 'shared:execute:4', 'check:execute:4', 'after:execute:4'],
+            $this->getExecutions()
+        );
+    }
+
+    /**
+     * A blocking task that received items before a stop is proceeded: its output goes on through the task that has
+     * stopped (it was dropped).
+     */
+    public function testABlockingTaskProceededAfterAStopGoesOnThroughTheTaskThatHasStopped(): void
+    {
+        $this->execute([
+            'items' => ['type' => 'iterable', 'items' => [1, 2, 3], 'outputs' => ['aggregate', 'check']],
+            'aggregate' => ['type' => 'blocking', 'outputs' => ['check']],
+            'check' => ['stop_on' => 2, 'outputs' => ['after']],
+            'after' => [],
+        ]);
+
+        self::assertSame(
+            ['aggregate:execute:1', 'check:execute:1', 'after:execute:1', 'aggregate:execute:2', 'check:execute:2', 'aggregate:proceed', 'check:execute:1,2', 'after:execute:1,2'],
+            \array_slice($this->events, 0, 8)
+        );
+    }
+
+    /**
+     * An iterable task whose iteration has been interrupted by a stop stays stopped: it would resume its interrupted
+     * iteration.
+     */
+    public function testAnIterableTaskInterruptedByAStopStaysStopped(): void
+    {
+        $this->execute([
+            'first' => ['type' => 'iterable', 'items' => [1], 'outputs' => ['inner']],
+            'second' => ['type' => 'iterable', 'items' => [2], 'outputs' => ['inner']],
+            'inner' => ['type' => 'iterable', 'items' => [10, 20], 'outputs' => ['check']],
+            'check' => ['stop_on' => 10],
+        ]);
+
+        self::assertSame(['check:execute:10'], $this->getExecutions());
+    }
+
+    /**
+     * A task whose initialization has failed (deprecated) stays stopped: none of its executions goes on.
+     */
+    public function testATaskWhoseInitializationHasFailedStaysStopped(): void
+    {
+        set_error_handler(static fn (): bool => true, \E_USER_DEPRECATED);
+        try {
+            $this->execute([
+                'items' => ['type' => 'iterable', 'items' => [1, 2], 'outputs' => ['bad']],
+                'bad' => ['type' => 'failed_initialization', 'outputs' => ['after']],
+                'after' => [],
+            ]);
+        } finally {
+            restore_error_handler();
+        }
+
+        self::assertSame(['bad:execute:1'], $this->getExecutions());
+    }
+
     /**
      * Once an iterable task has no more item, its next tasks and their error outputs are flushed, down to the
      * blocking tasks. (Each task is flushed again with its next tasks once resolved: only the first flushes are
@@ -525,6 +600,22 @@ public function proceed(ProcessState $state): void
                     $this->inputs = [];
                 }
             },
+            // Its initialization fails, not its executions
+            'failed_initialization' => new class($execute) implements InitializableTaskInterface {
+                public function __construct(private readonly \Closure $execute)
+                {
+                }
+
+                public function initialize(ProcessState $state): void
+                {
+                    throw new \InvalidArgumentException('Invalid configuration');
+                }
+
+                public function execute(ProcessState $state): void
+                {
+                    ($this->execute)($state);
+                }
+            },
             default => new class($execute) implements TaskInterface {
                 public function __construct(private readonly \Closure $execute)
                 {