#373 - Finalize the initialized tasks when the process fails

CHANGELOG.md

@@ -7,6 +7,7 @@ Latest
 * [#369](https://github.com/cleverage/process-bundle/issues/369) `StopTask` (`stop`): a process that has already failed (e.g. after a task initialization failure, going on being deprecated) stays failed instead of being flagged as stopped: the `Process <code> failed` log is written at the `failed_level` and `ProcessEvent::isProcessStopped()` is `false`, as before #337. Update documentation, add tests.
 * [#370](https://github.com/cleverage/process-bundle/issues/370) `RulesTransformer` (`rules`): the result of a `condition` is cast to `bool`, so a condition returning an `int` (`preg_match()`), a string, `null` or any other non-boolean value (a single variable) no longer throws a `TypeError`: the rule matches if the result is truthy. Update documentation, add tests.
 * [#372](https://github.com/cleverage/process-bundle/issues/372) `ProcessConfigurationRegistry`: an unknown task in the `outputs` or `error_outputs` of a task, or in the `entry_point` or `end_point` of a process, throws an `InvalidProcessConfigurationException` naming the process, the task and the key holding the reference, instead of a `MissingTaskConfigurationException` (`No task configuration with code : <code>`). A process whose configuration is not valid is left out of `getProcessConfigurations()` (`cleverage:process:list`, lists of the UI) with an error log on the `cleverage_process` channel, instead of making every listing fail; getting or executing this process still throws. Update documentation, add tests.
+* [#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
 -----

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.

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).

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;

docs/reference/tasks/stat_counter_task.md

@@ -20,7 +20,7 @@ Possible outputs
 
 `any`: the input, unchanged.
 
-At finalization, the message `Processed item count: <count>` is logged.
+At finalization, the message `Processed item count: <count>` is logged, also when the process fails.
 
 Options
 -------

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;