Skip to content

#333 - ProcessManager: restore the previous task configuration after a child task or a sub-process #334

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
njoubert-cleverage merged 2 commits into from
Oct 23, 2026
Merged

#333 - ProcessManager: restore the previous task configuration after a child task or a sub-process #334

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
b4ef451
#333 - ProcessManager: restore the previous task configuration when a…
njoubert-cleverage Oct 22, 2026
b23f532
Merge branch 'main' into 333
njoubert-cleverage Oct 23, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions CHANGELOG.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -4,6 +4,7 @@ Latest
## Fixes
* [#343](https://github.com/cleverage/process-bundle/issues/343) The temporary files of `FileSplitterTask` (`file_splitter`), `CsvSplitterTask` (`csv_splitter`) and of the JSON buffer of `ProcessLauncherTask` (`json_buffering`) are created with a random name, exclusively (an existing path, symbolic links included, is never reused) and readable by their owner only, instead of a name derived from the time (`uniqid()`), opened in `w` mode with the permissions of the umask. Update documentation, add tests.
* [#331](https://github.com/cleverage/process-bundle/issues/331) `ProcessManager`: the iterable tasks that produced items and the blocking tasks waiting for `proceed()` are tracked by task configuration instead of task code, and forgotten when their process ends. A task of a sub-process (`process_executor`) with the same code as a task of the parent no longer makes the parent lose the output of its blocking task (without error) or skip the flush at the end of its iteration (merged batches), and a failed sub-process no longer leaves its tasks to its next execution. Add tests.
* [#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
-----
Expand Down
138 changes: 76 additions & 62 deletions src/Manager/ProcessManager.php
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -132,6 +132,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),
Expand Down Expand Up @@ -160,6 +161,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;
Expand Down Expand Up @@ -289,6 +292,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()
Expand Down Expand Up @@ -323,90 +327,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
Expand Down Expand Up @@ -521,6 +534,7 @@ protected function finalize(TaskConfiguration $taskConfiguration): void
{
$task = $taskConfiguration->getTask();
if ($task instanceof FinalizableTaskInterface) {
$previousTaskConfiguration = $this->taskConfiguration;
$this->taskConfiguration = $taskConfiguration;
$state = $taskConfiguration->getState();
try {
Expand All @@ -533,7 +547,7 @@ protected function finalize(TaskConfiguration $taskConfiguration): void
$state->stop($e);
}
$this->handleState($state);
$this->taskConfiguration = null;
$this->taskConfiguration = $previousTaskConfiguration;
}
}

Expand Down
Loading
Loading