From b4ef451fbccbcb1b85851d2b1f998ff8b8f515fe Mon Sep 17 00:00:00 2001 From: Nicolas Joubert Date: Thu, 22 Oct 2026 17:12:17 +0200 Subject: [PATCH] #333 - ProcessManager: restore the previous task configuration when a task, an initialization, a finalization or a sub-process ends, so that the logs of an iterable task keep their task after its first item and the logs of a parent process are no longer attributed to a task of its sub-process. Add tests. Co-Authored-By: Claude Opus 5.5 (1M context) --- CHANGELOG.md | 3 + src/Manager/ProcessManager.php | 138 +++++----- .../ProcessManagerTaskConfigurationTest.php | 247 ++++++++++++++++++ 3 files changed, 326 insertions(+), 62 deletions(-) create mode 100644 tests/Manager/ProcessManagerTaskConfigurationTest.php diff --git a/CHANGELOG.md b/CHANGELOG.md index 9b33c604..51b84c52 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,6 +1,9 @@ Latest ------ +## Fixes +* [#333](https://github.com/cleverage/process-bundle/issues/333) `ProcessManager`: restore the previous task configuration (`getTaskConfiguration()`, read by the log processors for `task_code` and `task_service`) when a task, an initialization, a finalization or a sub-process ends. The logs of an iterable task no longer lose their task after its first item, and the logs of a parent process are no longer attributed to a task of its sub-process; the failing task is kept for the logs of the process failure. Add tests. + v5.4.1 ----- diff --git a/src/Manager/ProcessManager.php b/src/Manager/ProcessManager.php index bf5e7d96..adf2af37 100644 --- a/src/Manager/ProcessManager.php +++ b/src/Manager/ProcessManager.php @@ -128,6 +128,7 @@ public function getTaskConfiguration(): ?TaskConfiguration public function execute(string $processCode, mixed $input = null, array $context = []): mixed { $parentProcessHistory = $this->processHistory; + $parentTaskConfiguration = $this->taskConfiguration; try { $this->eventDispatcher->dispatch( new ProcessEvent($processCode, $input, $context), @@ -156,6 +157,8 @@ public function execute(string $processCode, mixed $input = null, array $context // When the process has failed, doExecute() did not restore the history of the parent process (if any): the // logs above are still attributed to the failed process, the next ones to the parent process $this->processHistory = $parentProcessHistory; + // The logs of the parent process are attributed to the task that has executed this process (if any) + $this->taskConfiguration = $parentTaskConfiguration; } return $result; @@ -279,6 +282,7 @@ protected function resolve(TaskConfiguration $taskConfiguration): bool */ protected function initialize(TaskConfiguration $taskConfiguration): void { + $previousTaskConfiguration = $this->taskConfiguration; $this->taskConfiguration = $taskConfiguration; if (TaskConfiguration::STRATEGY_STOP === $taskConfiguration->getErrorStrategy() @@ -313,90 +317,99 @@ protected function initialize(TaskConfiguration $taskConfiguration): void } $this->handleState($taskConfiguration->getState()); - $this->taskConfiguration = null; + $this->taskConfiguration = $previousTaskConfiguration; } protected function process(TaskConfiguration $taskConfiguration, int $executionFlag = self::EXECUTE_PROCESS): void { + // Restored when the task ends: the parent task (if any) goes on with its next items + $previousTaskConfiguration = $this->taskConfiguration; $this->taskConfiguration = $taskConfiguration; - $state = $taskConfiguration->getState(); - do { - // Execute the current task, and fetch status - $this->processExecution($taskConfiguration, $executionFlag); + try { + $state = $taskConfiguration->getState(); + do { + // Execute the current task, and fetch status + $this->processExecution($taskConfiguration, $executionFlag); - $this->handleState($state); + $this->handleState($state); - // An error feed cannot be blocked or skipped (even if the process has been stopped) - if ($state->hasErrorOutput()) { - foreach ($taskConfiguration->getErrorTasksConfigurations() as $errorTask) { - $this->prepareNextProcess($taskConfiguration, $errorTask, true); - $this->process($errorTask); + // An error feed cannot be blocked or skipped (even if the process has been stopped) + if ($state->hasErrorOutput()) { + foreach ($taskConfiguration->getErrorTasksConfigurations() as $errorTask) { + $this->prepareNextProcess($taskConfiguration, $errorTask, true); + $this->process($errorTask); - // Bubble up the stop signal - if ($errorTask->getState()->isStopped()) { - $state->setStopped(true); + // Bubble up the stop signal + if ($errorTask->getState()->isStopped()) { + $state->setStopped(true); - return; + return; + } } } - } - if ($state->isStopped()) { - $exception = $state->getException(); - if ($exception instanceof \Throwable) { - throw ProcessFailedException::create($state->getProcessConfiguration()->getCode(), $state->getTaskConfiguration()->getCode(), $exception); - } + if ($state->isStopped()) { + $exception = $state->getException(); + if ($exception instanceof \Throwable) { + throw ProcessFailedException::create($state->getProcessConfiguration()->getCode(), $state->getTaskConfiguration()->getCode(), $exception); + } - return; - } + return; + } - // Run child items only if the state is not "skipped" and task is not blocking - $task = $taskConfiguration->getTask(); - $shouldContinue = - (!$task instanceof BlockingTaskInterface || self::EXECUTE_PROCEED === $executionFlag) - && !$state->isSkipped(); + // Run child items only if the state is not "skipped" and task is not blocking + $task = $taskConfiguration->getTask(); + $shouldContinue = + (!$task instanceof BlockingTaskInterface || self::EXECUTE_PROCEED === $executionFlag) + && !$state->isSkipped(); - if ($shouldContinue) { - if ($task instanceof IterableTaskInterface) { - // Register the task as not empty - $this->addProcessedIterable($taskConfiguration); - } - foreach ($taskConfiguration->getNextTasksConfigurations() as $nextTaskConfiguration) { - $this->prepareNextProcess($taskConfiguration, $nextTaskConfiguration); - $this->process($nextTaskConfiguration); + if ($shouldContinue) { + if ($task instanceof IterableTaskInterface) { + // Register the task as not empty + $this->addProcessedIterable($taskConfiguration); + } + foreach ($taskConfiguration->getNextTasksConfigurations() as $nextTaskConfiguration) { + $this->prepareNextProcess($taskConfiguration, $nextTaskConfiguration); + $this->process($nextTaskConfiguration); - // Bubble up the stop signal - if ($nextTaskConfiguration->getState()->isStopped()) { - $state->setStopped(true); + // Bubble up the stop signal + if ($nextTaskConfiguration->getState()->isStopped()) { + $state->setStopped(true); - return; + return; + } } + $state->setSkipped(false); // Reset skipped state } - $state->setSkipped(false); // Reset skipped state - } - $hasMoreItem = false; // By default, a task is not iterable - if ($task instanceof IterableTaskInterface) { - // Check if task has more items - $hasMoreItem = $task->next($state); - if (!$hasMoreItem) { - if (!$this->hasProcessedIterable($taskConfiguration)) { - return; // This means the task is empty - } - // This means we are over iterating this task so we can remove it from registry - $this->removeProcessedIterable($taskConfiguration); - if (self::EXECUTE_FLUSH !== $executionFlag) { - // This task is now finished, we may flush it to test if there is anything lasting - $this->flush($taskConfiguration); - } - if ($state->isStopped()) { - return; + $hasMoreItem = false; // By default, a task is not iterable + if ($task instanceof IterableTaskInterface) { + // Check if task has more items + $hasMoreItem = $task->next($state); + if (!$hasMoreItem) { + if (!$this->hasProcessedIterable($taskConfiguration)) { + return; // This means the task is empty + } + // This means we are over iterating this task so we can remove it from registry + $this->removeProcessedIterable($taskConfiguration); + if (self::EXECUTE_FLUSH !== $executionFlag) { + // This task is now finished, we may flush it to test if there is anything lasting + $this->flush($taskConfiguration); + } + if ($state->isStopped()) { + return; + } } } - } - } while ($hasMoreItem); + } while ($hasMoreItem); + } catch (\Throwable $error) { + // Keep the failing task for the logs of the failure, execute() restores the task of the parent process + $previousTaskConfiguration = $this->taskConfiguration; - $this->taskConfiguration = null; + throw $error; + } finally { + $this->taskConfiguration = $previousTaskConfiguration; + } } protected function processExecution(TaskConfiguration $taskConfiguration, int $executionFlag): void @@ -511,6 +524,7 @@ protected function finalize(TaskConfiguration $taskConfiguration): void { $task = $taskConfiguration->getTask(); if ($task instanceof FinalizableTaskInterface) { + $previousTaskConfiguration = $this->taskConfiguration; $this->taskConfiguration = $taskConfiguration; $state = $taskConfiguration->getState(); try { @@ -523,7 +537,7 @@ protected function finalize(TaskConfiguration $taskConfiguration): void $state->stop($e); } $this->handleState($state); - $this->taskConfiguration = null; + $this->taskConfiguration = $previousTaskConfiguration; } } diff --git a/tests/Manager/ProcessManagerTaskConfigurationTest.php b/tests/Manager/ProcessManagerTaskConfigurationTest.php new file mode 100644 index 00000000..045e2a5b --- /dev/null +++ b/tests/Manager/ProcessManagerTaskConfigurationTest.php @@ -0,0 +1,247 @@ + + */ + private array $observed = []; + + protected function setUp(): void + { + $container = $this->container = new Container(); + // Emits 2 items, and records the current task configuration each time + $container->set('test.iterate', new class($this) implements IterableTaskInterface { + private int $index = 0; + + public function __construct( + private readonly ProcessManagerTaskConfigurationTest $test, + ) { + } + + public function execute(ProcessState $state): void + { + $this->test->observe(); + $state->setOutput($this->index); + } + + public function next(ProcessState $state): bool + { + return ++$this->index < 2; + } + }); + $container->set('test.noop', new class implements TaskInterface { + public function execute(ProcessState $state): void + { + } + }); + $container->set('test.failing', new class implements TaskInterface { + public function execute(ProcessState $state): void + { + throw new \LogicException('Something went wrong'); + } + }); + // Runs the given sub-process (ignoring its failure, as the process_executor task with the skip error + // strategy), then records the current task configuration + $container->set('test.sub_process', new class($this) implements TaskInterface { + public function __construct( + private readonly ProcessManagerTaskConfigurationTest $test, + ) { + } + + public function execute(ProcessState $state): void + { + $this->test->executeSubProcess((string) $state->getTaskConfiguration()->getOption('process')); + $this->test->observe(); + } + }); + + $this->registry = new ProcessConfigurationRegistry( + [ + 'test.iterate' => $this->createProcessConfiguration([ + 'iterate' => $this->createTaskConfiguration('@test.iterate', [], ['child']), + 'child' => $this->createTaskConfiguration('@test.noop'), + ]), + 'test.parent' => $this->createProcessConfiguration([ + 'sub_process' => $this->createTaskConfiguration('@test.sub_process', ['process' => 'test.child']), + ]), + 'test.child' => $this->createProcessConfiguration([ + 'child_task' => $this->createTaskConfiguration('@test.noop'), + ]), + 'test.failing_parent' => $this->createProcessConfiguration([ + 'sub_process' => $this->createTaskConfiguration('@test.sub_process', ['process' => 'test.failing']), + ]), + 'test.failing' => $this->createProcessConfiguration([ + 'fail' => $this->createTaskConfiguration('@test.failing'), + ]), + ], + 'stop' + ); + + $this->processManager = $this->createProcessManager(new EventDispatcher()); + } + + public function observe(): void + { + $this->observed[] = $this->processManager->getTaskConfiguration()?->getCode(); + } + + public function executeSubProcess(string $processCode): void + { + try { + $this->processManager->execute($processCode); + } catch (ProcessFailedException) { + } + } + + public function testTheIterableTaskIsRestoredAfterItsChildTask(): void + { + $this->processManager->execute('test.iterate'); + + self::assertSame(['iterate', 'iterate'], $this->observed); + self::assertNull($this->processManager->getTaskConfiguration()); + } + + public function testTheTaskIsRestoredAfterASubProcess(): void + { + $this->processManager->execute('test.parent'); + + self::assertSame(['sub_process'], $this->observed); + self::assertNull($this->processManager->getTaskConfiguration()); + } + + public function testTheTaskIsRestoredAfterAFailedSubProcess(): void + { + $this->processManager->execute('test.failing_parent'); + + self::assertSame(['sub_process'], $this->observed); + self::assertNull($this->processManager->getTaskConfiguration()); + } + + public function testTheFailingTaskIsKeptForTheLogsOfTheFailure(): void + { + $failingTask = null; + $dispatcher = new EventDispatcher(); + $dispatcher->addListener(ProcessEvent::EVENT_PROCESS_FAILED, function () use (&$failingTask): void { + $failingTask = $this->processManager->getTaskConfiguration()?->getCode(); + }); + $this->processManager = $this->createProcessManager($dispatcher); + + try { + $this->processManager->execute('test.failing'); + self::fail('The process should have failed'); + } catch (ProcessFailedException) { + } + + self::assertSame('fail', $failingTask); + self::assertNull($this->processManager->getTaskConfiguration()); + } + + private function createProcessManager(EventDispatcher $eventDispatcher): ProcessManager + { + return new ProcessManager( + $this->container, + new ProcessLogger(new NullLogger()), + new TaskLogger(new NullLogger()), + $this->registry, + new ContextualOptionResolver(), + $eventDispatcher, + taskRegistry: new TaskRegistry($this->container), + ); + } + + /** + * @param array> $tasks + * + * @return array + */ + private function createProcessConfiguration(array $tasks): array + { + return [ + 'options' => [], + 'entry_point' => null, + 'end_point' => null, + 'description' => '', + 'help' => '', + 'public' => true, + 'tasks' => $tasks, + ]; + } + + /** + * @param array $options + * @param list $outputs + * + * @return array + */ + 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, + ]; + } +}