From 2c65a214798cc168f389da66ee9c1a6b4170c741 Mon Sep 17 00:00:00 2001 From: Nicolas Joubert Date: Fri, 6 Nov 2026 14:13:57 +0100 Subject: [PATCH] #481 - A stop only stops the flow it happened in Co-Authored-By: Claude Opus 5.5 --- CHANGELOG.md | 1 + UPGRADE.md | 7 ++ docs/04-advanced_workflow.md | 4 +- src/Manager/ProcessManager.php | 22 ++++++ tests/Manager/ProcessManagerFlowTest.php | 91 ++++++++++++++++++++++++ 5 files changed, 124 insertions(+), 1 deletion(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 9ff28fda..dffe4287 100644 --- a/CHANGELOG.md +++ b/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. diff --git a/UPGRADE.md b/UPGRADE.md index 63f506c4..145fd79e 100644 --- a/UPGRADE.md +++ b/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 diff --git a/docs/04-advanced_workflow.md b/docs/04-advanced_workflow.md index 89d2cd2a..a090008a 100644 --- a/docs/04-advanced_workflow.md +++ b/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 diff --git a/src/Manager/ProcessManager.php b/src/Manager/ProcessManager.php index b22e4ade..738af2f6 100644 --- a/src/Manager/ProcessManager.php +++ b/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 + */ + 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); diff --git a/tests/Manager/ProcessManagerFlowTest.php b/tests/Manager/ProcessManagerFlowTest.php index c27329b3..bfad3487 100644 --- a/tests/Manager/ProcessManagerFlowTest.php +++ b/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) {