From 04f43cbf9d77b5af8eeedbed0834c232bf979ec4 Mon Sep 17 00:00:00 2001 From: Nicolas Joubert Date: Fri, 23 Oct 2026 16:08:29 +0200 Subject: [PATCH] #373 - Finalize the initialized tasks when the process fails Co-Authored-By: Claude Opus 5.5 (1M context) --- CHANGELOG.md | 3 + UPGRADE.md | 13 + docs/02-task_types.md | 19 +- docs/04-advanced_workflow.md | 6 +- docs/reference/tasks/stat_counter_task.md | 2 +- src/Manager/ProcessManager.php | 15 + .../ProcessManagerFinalizationTest.php | 402 ++++++++++++++++++ 7 files changed, 453 insertions(+), 7 deletions(-) create mode 100644 tests/Manager/ProcessManagerFinalizationTest.php diff --git a/CHANGELOG.md b/CHANGELOG.md index 9e55a44b..245f5c03 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,6 +1,9 @@ Latest ------ +## Fixes +* [#373](https://github.com/cleverage/process-bundle/issues/373) `ProcessManager`: the tasks are finalized (`FinalizableTaskInterface::finalize()`) when the process fails too, so that a long-running worker (Messenger, UI bundle) no longer keeps the resources of a failed process open (files of the CSV, JSON stream and XML tasks, cursor of `doctrine_database_reader` blocking the next queries with MySQL unbuffered queries). Only the initialized tasks are finalized, each one once; a finalization failure is logged as critical and the process fails with its original exception. `StatCounterTask` (`stat_counter`) logs its count when the process fails too. Update documentation, add tests. + v5.5 ----- diff --git a/UPGRADE.md b/UPGRADE.md index 1cd439a9..aa0a4139 100644 --- a/UPGRADE.md +++ b/UPGRADE.md @@ -1,6 +1,19 @@ Upgrade Guide ============= +## v5.6 + +Some behaviours change: check the points below. + +### Finalization of the tasks when the process fails + +`FinalizableTaskInterface::finalize()` is now also called when the process fails, on every task that has been +initialized (not on a task whose `initialize()` failed, nor on the tasks not reached). A custom finalizable task must not +assume that the process succeeded nor that its data are complete: only release its resources (close files, free +cursors), do not write a footer, move / publish a partial file or run long operations. Check +`$state->getProcessHistory()->isFailed()` in `finalize()` to skip such work when the process has failed. +`StatCounterTask` (`stat_counter`) now logs its count when the process fails too. + ## v5.5 No deprecation in this version. Some behaviours change: check the points below. diff --git a/docs/02-task_types.md b/docs/02-task_types.md index fc40b2c4..2578e90d 100644 --- a/docs/02-task_types.md +++ b/docs/02-task_types.md @@ -24,7 +24,7 @@ A task is *resolved* when all its executions are over, as well as those of all i | `BlockingTaskInterface` | `proceed(ProcessState $state): void` | Once, when all parent tasks are resolved, only if the task received at least one input | | `FlushableTaskInterface` | `flush(ProcessState $state): void` | When an ancestor (or the task itself) is resolved or finishes iterating: possibly several times | | `InitializableTaskInterface` | `initialize(ProcessState $state): void` | Once, before the process starts | -| `FinalizableTaskInterface` | `finalize(ProcessState $state): void` | Once, at the end of the process | +| `FinalizableTaskInterface` | `finalize(ProcessState $state): void` | Once, at the end of the process, whether it succeeds or fails | ## Iterable tasks @@ -162,9 +162,20 @@ temporary folder, log some statistics). Finalizable tasks implement `CleverAge\ProcessBundle\Model\FinalizableTaskInterface`. Their `finalize` method is called once all tasks are resolved, for every task of the process, in the order they are configured. -Note that finalization only happens when the process reaches its end: if a task stops the process because of an -exception (error strategy `stop`), the process fails immediately and `finalize` is not called. A process stopped -without exception (e.g. by the [StopTask](reference/tasks/stop_task.md)) is still finalized. +A process stopped without exception (e.g. by the [StopTask](reference/tasks/stop_task.md)) is finalized the same way. + +`finalize` is also called when the process fails (e.g. a task stops the process because of an exception with the error +strategy `stop`), so that the resources are released (files, database cursors...), e.g. for the next executions of a +long-running worker (Messenger, UI). In this case: +* only the tasks that have been initialized are finalized: not the tasks whose `initialize` has failed, nor the tasks + that were not reached because the process failed during the initialization; +* the tasks are finalized after the failure, in the order they are configured: `finalize` must not assume that the + data are complete (e.g. a writer must only close its file, not write a footer or move a partial file to its + destination), nor take long: `$state->getProcessHistory()->isFailed()` tells whether the process has failed; +* an exception thrown by `finalize` is logged as critical and the next tasks are still finalized: the process fails + with its original exception. + +Each task is finalized only once: a task already finalized when the process fails is not finalized again. Examples of finalizable tasks: [CsvReaderTask](reference/tasks/csv_reader_task.md) and the other CSV tasks (close the file), [StatCounterTask](reference/tasks/stat_counter_task.md) (logs the number of processed items). diff --git a/docs/04-advanced_workflow.md b/docs/04-advanced_workflow.md index 24d6e848..12fe1f73 100644 --- a/docs/04-advanced_workflow.md +++ b/docs/04-advanced_workflow.md @@ -23,7 +23,9 @@ When a process is executed (with the `cleverage:process:execute` command or with `cleverage_process.end` event (`ProcessEvent::isProcessStopped()` tells whether the process was stopped on purpose by a [StopTask](reference/tasks/stop_task.md)) -If an exception is thrown at any point, the `cleverage_process.fail` event is dispatched and the exception is rethrown. +If an exception is thrown at any point, the tasks that have been initialized (and not finalized yet) are finalized, +the `cleverage_process.fail` event is dispatched and the exception is rethrown (see +[finalizable tasks](02-task_types.md#finalizable-tasks)). Note that a task error handled by the `stop` strategy does not surface as the original exception: the process manager throws a `CleverAge\ProcessBundle\Exception\ProcessFailedException` (a `\RuntimeException`), whose message contains the process code, the task code and the original message. The original exception is available with `getPrevious()`. @@ -262,7 +264,7 @@ When a checker returns `true`, the task is not executed and the process is stopp `stop` error strategy, whatever its `error_strategy` (its error outputs are not executed): the process fails with a `ProcessFailedException` whose previous exception is a `CleverAge\ProcessBundle\Exception\ProcessStoppedException`, and the `cleverage_process.fail` event is dispatched. As for any failure, the remaining items are not flushed and the -tasks are not finalized. +initialized tasks are finalized. ```php namespace App\Process; diff --git a/docs/reference/tasks/stat_counter_task.md b/docs/reference/tasks/stat_counter_task.md index 70b002f3..f6f115df 100644 --- a/docs/reference/tasks/stat_counter_task.md +++ b/docs/reference/tasks/stat_counter_task.md @@ -20,7 +20,7 @@ Possible outputs `any`: the input, unchanged. -At finalization, the message `Processed item count: ` is logged. +At finalization, the message `Processed item count: ` is logged, also when the process fails. Options ------- diff --git a/src/Manager/ProcessManager.php b/src/Manager/ProcessManager.php index 02ad208f..7fa9c332 100644 --- a/src/Manager/ProcessManager.php +++ b/src/Manager/ProcessManager.php @@ -183,12 +183,18 @@ protected function doExecute(string $processCode, mixed $input = null, array $co $processHistory = $this->initializeStates($processConfiguration, $context); $this->processHistory = $processHistory; + // Finalized if the process fails: the initialized tasks that are not finalized yet (indexed by spl_object_id()) + $tasksToFinalize = []; try { $this->checkProcess($processConfiguration); // First initialize the whole stack in a linear way, tasks are initialized in the order they are configured foreach ($processConfiguration->getTaskConfigurations() as $taskConfiguration) { $this->initialize($taskConfiguration); + // A task whose initialization has failed is stopped: it is not finalized if the process fails + if (!$taskConfiguration->getState()->isStopped()) { + $tasksToFinalize[spl_object_id($taskConfiguration)] = $taskConfiguration; + } } // If defined, set the input of a task @@ -211,6 +217,8 @@ protected function doExecute(string $processCode, mixed $input = null, array $co // Finalize the process in a linear way foreach ($processConfiguration->getTaskConfigurations() as $taskConfiguration) { + // Never finalized twice, even if the process fails after this point + unset($tasksToFinalize[spl_object_id($taskConfiguration)]); $this->finalize($taskConfiguration); } @@ -224,9 +232,16 @@ protected function doExecute(string $processCode, mixed $input = null, array $co ->getOutput(); } } catch (\Throwable $error) { + // Before the finalization: the tasks can tell the process has failed (ProcessHistory::isFailed()) if ($processHistory->isStarted() || $processHistory->isStopped()) { $processHistory->setFailed(); } + + // Release the resources of the tasks (files, database cursors...), e.g. for the next executions of a + // long-running worker. A finalization failure is logged by finalize(), and does not hide the original error + foreach ($tasksToFinalize as $taskConfiguration) { + $this->finalize($taskConfiguration); + } $this->endProcess($processHistory); throw $error; diff --git a/tests/Manager/ProcessManagerFinalizationTest.php b/tests/Manager/ProcessManagerFinalizationTest.php new file mode 100644 index 00000000..93f3b505 --- /dev/null +++ b/tests/Manager/ProcessManagerFinalizationTest.php @@ -0,0 +1,402 @@ + + */ + private array $finalized = []; + + /** + * Whether the process history was failed during each finalization. + * + * @var list + */ + private array $failedAtFinalization = []; + + /** + * The logs of the tasks, with the task configuration seen by the logs. + * + * @var list + */ + private array $taskLogs = []; + + /** + * A message of the process logs that throws an exception when logged. + */ + private ?string $failingProcessLog = null; + + protected function setUp(): void + { + $this->container = new Container(); + $this->container->set('test.noop', new class implements TaskInterface { + public function execute(ProcessState $state): void + { + } + }); + $this->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, as the process_executor task + $this->container->set('test.sub_process', new class($this) implements TaskInterface { + public function __construct( + private readonly ProcessManagerFinalizationTest $test, + ) { + } + + public function execute(ProcessState $state): void + { + $this->test->executeSubProcess((string) $state->getTaskConfiguration()->getOption('process')); + } + }); + foreach (['first', 'second', 'third', 'child'] as $code) { + $this->container->set("test.finalizable_{$code}", $this->createFinalizableTask($code)); + } + $this->container->set('test.finalizable_throwing', $this->createFinalizableTask('throwing', failFinalization: true)); + $this->container->set('test.finalizable_bad_init', $this->createFinalizableTask('bad_init', failInitialization: true)); + + $this->eventDispatcher = new EventDispatcher(); + } + + public function executeSubProcess(string $processCode): void + { + $this->processManager->execute($processCode); + } + + public function testTheTasksAreFinalizedWhenTheProcessSucceeds(): void + { + $this->execute([ + 'first' => $this->createTaskConfiguration('@test.finalizable_first', ['second']), + 'second' => $this->createTaskConfiguration('@test.finalizable_second'), + ]); + + self::assertSame(['first (first)', 'second (second)'], $this->finalized); + self::assertSame([false, false], $this->failedAtFinalization); + self::assertNull($this->processManager->getTaskConfiguration()); + } + + public function testTheTasksAreFinalizedWhenTheProcessFails(): void + { + $failingTask = null; + $this->eventDispatcher->addListener(ProcessEvent::EVENT_PROCESS_FAILED, function () use (&$failingTask): void { + $failingTask = $this->processManager->getTaskConfiguration()?->getCode(); + }); + + $error = $this->executeFailing([ + 'first' => $this->createTaskConfiguration('@test.finalizable_first', ['fail']), + 'fail' => $this->createTaskConfiguration('@test.failing', ['second']), + // Initialized, never executed + 'second' => $this->createTaskConfiguration('@test.finalizable_second'), + ]); + + self::assertInstanceOf(ProcessFailedException::class, $error); + self::assertSame('Something went wrong', $error->getPrevious()?->getMessage()); + // The logs of each finalization are attributed to the finalized task + self::assertSame(['first (first)', 'second (second)'], $this->finalized); + // The tasks can tell the process has failed + self::assertSame([true, true], $this->failedAtFinalization); + // The failing task is kept for the logs of the failure + self::assertSame('fail', $failingTask); + self::assertNull($this->processManager->getTaskConfiguration()); + } + + public function testATaskWhoseInitializationHasFailedIsNotFinalizedWhenTheProcessFails(): void + { + set_error_handler(static fn (): bool => true, \E_USER_DEPRECATED); + try { + $error = $this->executeFailing([ + 'first' => $this->createTaskConfiguration('@test.finalizable_first', ['bad_init']), + 'bad_init' => $this->createTaskConfiguration('@test.finalizable_bad_init'), + ]); + } finally { + restore_error_handler(); + } + + self::assertInstanceOf(ProcessFailedException::class, $error); + self::assertSame('Invalid configuration', $error->getPrevious()?->getMessage()); + self::assertSame(['first (first)'], $this->finalized); + } + + public function testTheTasksThatAreNotInitializedAreNotFinalizedWhenTheProcessFails(): void + { + $error = $this->executeFailing([ + 'first' => $this->createTaskConfiguration('@test.finalizable_first', ['missing']), + 'missing' => $this->createTaskConfiguration('test.missing', ['second']), + 'second' => $this->createTaskConfiguration('@test.finalizable_second'), + ]); + + self::assertInstanceOf(UnexpectedValueException::class, $error); + self::assertSame(['first (first)'], $this->finalized); + // Also failed for an exception thrown outside a task + self::assertSame([true], $this->failedAtFinalization); + } + + public function testAFinalizationFailureIsLoggedAndDoesNotHideTheErrorOfTheProcess(): void + { + $error = $this->executeFailing([ + 'throwing' => $this->createTaskConfiguration('@test.finalizable_throwing', ['fail']), + 'fail' => $this->createTaskConfiguration('@test.failing'), + 'second' => $this->createTaskConfiguration('@test.finalizable_second'), + ]); + + self::assertInstanceOf(ProcessFailedException::class, $error); + self::assertSame('Something went wrong', $error->getPrevious()?->getMessage()); + // The next tasks are still finalized + self::assertSame(['throwing (throwing)', 'second (second)'], $this->finalized); + self::assertContains(['critical', 'Finalization failed', 'throwing'], $this->taskLogs); + } + + public function testTheTasksAreNotFinalizedTwiceWhenTheProcessFailsAfterTheirFinalization(): void + { + $this->failingProcessLog = 'Process test.process succeed'; + + $error = $this->executeFailing([ + 'first' => $this->createTaskConfiguration('@test.finalizable_first', ['second']), + 'second' => $this->createTaskConfiguration('@test.finalizable_second'), + ]); + + self::assertInstanceOf(\RuntimeException::class, $error); + self::assertSame('Logger failure', $error->getMessage()); + self::assertSame(['first (first)', 'second (second)'], $this->finalized); + } + + public function testAFailedSubProcessIsFinalizedAndItsParentGoesOn(): void + { + $this->createProcessManager([ + 'test.process' => [ + 'first' => $this->createTaskConfiguration('@test.finalizable_first', ['sub_process']), + 'sub_process' => $this->createTaskConfiguration('@test.sub_process', ['second'], ['process' => 'test.child'], 'skip'), + 'second' => $this->createTaskConfiguration('@test.finalizable_second'), + ], + 'test.child' => [ + 'child' => $this->createTaskConfiguration('@test.finalizable_child', ['fail']), + 'fail' => $this->createTaskConfiguration('@test.failing'), + ], + ]); + + $this->processManager->execute('test.process'); + + // The task of the sub-process is finalized when it fails, the tasks of the parent when it ends + self::assertSame(['child (child)', 'first (first)', 'second (second)'], $this->finalized); + self::assertNull($this->processManager->getTaskConfiguration()); + } + + public function record(string $code, bool $failed): void + { + $this->failedAtFinalization[] = $failed; + $this->finalized[] = "{$code} ({$this->processManager->getTaskConfiguration()?->getCode()})"; + } + + private function createFinalizableTask( + string $code, + bool $failFinalization = false, + bool $failInitialization = false, + ): TaskInterface { + return new class($this, $code, $failFinalization, $failInitialization) implements InitializableTaskInterface, FinalizableTaskInterface { + public function __construct( + private readonly ProcessManagerFinalizationTest $test, + private readonly string $code, + private readonly bool $failFinalization, + private readonly bool $failInitialization, + ) { + } + + public function initialize(ProcessState $state): void + { + if ($this->failInitialization) { + throw new \InvalidArgumentException('Invalid configuration'); + } + } + + /** + * Like configurable tasks (options are resolved again), fail again when executed. + */ + public function execute(ProcessState $state): void + { + $this->initialize($state); + $state->setOutput($state->getInput()); + } + + public function finalize(ProcessState $state): void + { + $this->test->record($this->code, $state->getProcessHistory()->isFailed()); + if ($this->failFinalization) { + throw new \RuntimeException('Finalization failed'); + } + } + }; + } + + /** + * @param array> $tasks + */ + private function execute(array $tasks): void + { + $this->createProcessManager(['test.process' => $tasks]); + $this->processManager->execute('test.process'); + } + + /** + * @param array> $tasks + */ + private function executeFailing(array $tasks): \Throwable + { + try { + $this->execute($tasks); + } catch (\Throwable $error) { + return $error; + } + + self::fail('The process should have failed'); + } + + /** + * @param array>> $processes + */ + private function createProcessManager(array $processes): void + { + $configurations = []; + foreach ($processes as $code => $tasks) { + $configurations[$code] = [ + 'options' => [], + 'entry_point' => null, + 'end_point' => null, + 'description' => '', + 'help' => '', + 'public' => true, + 'tasks' => $tasks, + ]; + } + + $test = $this; + $processLogger = new class($test) extends PsrAbstractLogger { + public function __construct( + private readonly ProcessManagerFinalizationTest $test, + ) { + } + + public function log($level, string|\Stringable $message, array $context = []): void + { + $this->test->logProcess((string) $message); + } + }; + $taskLogger = new class($test) extends PsrAbstractLogger { + public function __construct( + private readonly ProcessManagerFinalizationTest $test, + ) { + } + + public function log($level, string|\Stringable $message, array $context = []): void + { + $this->test->logTask((string) $level, (string) $message); + } + }; + + $this->processManager = new ProcessManager( + $this->container, + new ProcessLogger($processLogger), + new TaskLogger($taskLogger), + new ProcessConfigurationRegistry($configurations, 'stop'), + new ContextualOptionResolver(), + $this->eventDispatcher, + taskRegistry: new TaskRegistry($this->container), + ); + } + + public function logProcess(string $message): void + { + if ($message === $this->failingProcessLog) { + throw new \RuntimeException('Logger failure'); + } + } + + public function logTask(string $level, string $message): void + { + $this->taskLogs[] = [$level, $message, $this->processManager->getTaskConfiguration()?->getCode()]; + } + + /** + * @param list $outputs + * @param array $options + * + * @return array + */ + private function createTaskConfiguration( + string $service, + array $outputs = [], + array $options = [], + ?string $errorStrategy = null, + ): array { + return [ + 'service' => $service, + 'options' => $options, + 'description' => '', + 'help' => '', + 'outputs' => $outputs, + 'errors' => [], + 'error_outputs' => [], + 'error_strategy' => $errorStrategy, + 'log_level' => null, + ]; + } +}