Backport to 5.5 the fixes merged into main after 5.5 (#367, #368, #369, #370, #372, #373)

CHANGELOG.md

@@ -1,6 +1,14 @@
 Latest
 ------
 
+## Fixes
+* [#367](https://github.com/cleverage/process-bundle/issues/367) `ProcessLauncherTask` (`process_launcher`) with `json_buffering`: when the sub-process writes nothing into its JSON buffer (no `end_point`, or an output that is not an array), the task outputs nothing again, as before 5.5, instead of the path of an empty file, and the empty buffer is removed. The buffer of a failed sub-process, and those of the still running sub-processes stopped because of it, are removed too (one file was left in the temporary directory for each launch). Update documentation, add tests.
+* [#368](https://github.com/cleverage/process-bundle/issues/368) `YamlWriterTask` (`yaml_writer`): the parent directory is created for a local path only (no scheme, or `file://`), not for a stream wrapper (`compress.zlib://`, `s3://`...), which failed, or `php://stdout`, which created a `php:` directory. Update documentation, add tests.
+* [#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.5.1
+
+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.

config/services/registry.yaml

@@ -6,8 +6,10 @@ services:
             - ~
             - ~
             - ~
+            - '@logger' # Processes left out of the listings because their configuration is not valid
         tags:
             - { name: kernel.reset, method: reset }
+            - { name: monolog.logger, channel: cleverage_process }
     CleverAge\ProcessBundle\Registry\ProcessConfigurationRegistry:
         alias: cleverage_process.registry.process_configuration
     CleverAge\ProcessBundle\Registry\ProcessConfigurationRegistryInterface:

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;
@@ -316,7 +318,8 @@ PHP executes a process in a single thread. To use several CPU cores, the
   of its `end_point` is written to a JSON stream file whose path is outputted by the task once the sub-process is over
   (use a [JsonStreamReaderTask](reference/tasks/json_stream_reader_task.md) to read it back). The file is only written
   if the sub-process has an `end_point` whose last output is an array; otherwise, as when `json_buffering` is disabled,
-  the task outputs nothing
+  the task outputs nothing (and removes the empty file). The task does not remove the files it outputs: remove them
+  once read
 - if a sub-process fails (non-zero exit code), all the running sub-processes are stopped and the task throws an
   exception
 

docs/reference/01-process_definition.md

@@ -74,6 +74,18 @@ reads its `ui` key to configure the launch form.
 **tasks**: list of task definitions contained in the process, indexed by task code. See
 [task definition](02-task_definition.md).
 
+### Invalid task references
+
+The task codes referenced by the `entry_point` and the `end_point` of the process, and by the `outputs` and the
+`error_outputs` of its tasks, must be tasks of the process. They are not checked when the container is compiled, but
+when the process configuration is resolved: getting or executing the process throws an
+`InvalidProcessConfigurationException` naming the process, the unknown task and where it is referenced (e.g.
+`Unknown task 'missing' in the 'outputs' of the task 'read' (from process: app.import)`).
+
+Such a process, as any process whose configuration is not valid, is left out of the lists of processes
+(`cleverage:process:list`, `ProcessConfigurationRegistryInterface::getProcessConfigurations()` used e.g. by the UI),
+with an error log on the `cleverage_process` channel, instead of making them fail.
+
 ### Main branch
 
 Only the *main branch* of the process is executed: the group of connected tasks containing the `entry_point`, or else

docs/reference/tasks/process_launcher_task.md

@@ -26,8 +26,9 @@ Possible outputs
 
 Nothing is output when an input is received (the task is skipped). When `json_buffering` is enabled, the sub-process
 end point output (only if it is an array) is written to a JSON stream file of the system temporary directory (`<sys_get_temp_dir()>/cdm_buffer_*.json-stream`, in `var/` of the project before 5.4) and, once the sub-process is
-finished, the task outputs the path of this file (`string`), during the next iterations or when flushed. Without
-`json_buffering`, the task outputs nothing.
+finished, the task outputs the path of this file (`string`), during the next iterations or when flushed. If nothing
+was written (no `end_point`, or an output that is not an array), the empty file is removed and the task outputs
+nothing for this sub-process. Without `json_buffering`, the task outputs nothing.
 
 If a sub-process exits with a non-zero code, its error output is logged as critical, all running sub-processes are
 stopped and a `RuntimeException` is thrown.
@@ -69,5 +70,7 @@ Notes
 
 * Sub-processes are run with the console of the project (`%kernel.project_dir%/bin/console`) and the same Symfony
   environment as the current one (`--env`), the arguments of the service.
-* The JSON stream files are not removed by the task: remove them once read (e.g. with a [file remover](file_remover_task.md) task).
+* The JSON stream files output by the task are not removed by it: remove them once read (e.g. with a
+  [file remover](file_remover_task.md) task). The task only removes the files it does not output: empty ones, and
+  those of a failed sub-process and of the still running sub-processes stopped because of it.
 * The incremental error output (stderr) of the running sub-processes is echoed directly.

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

docs/reference/tasks/stop_task.md

@@ -13,6 +13,10 @@ file does not exist. Unlike [DieTask](die_task.md), the stop is handled by the p
 * `cleverage:process:execute` exits with code 0, and the task running a sub-process
   ([ProcessExecutorTask](process_executor_task.md)) goes on.
 
+A process that has already failed (e.g. a task whose initialization failed, see the
+[initializable tasks](../../02-task_types.md)) stays **failed**: the `StopTask` still stops the flow of the
+tasks, but it does not turn the failure into a stop.
+
 To stop a process because of an error, let the error fail the process instead (error strategy `stop`).
 
 Task reference

docs/reference/tasks/yaml_writer_task.md

@@ -24,7 +24,7 @@ Options
 
 | Code        | Type     | Required | Default | Description                                                            |
 |-------------|----------|:--------:|---------|------------------------------------------------------------------------|
-| `file_path` | `string` |  **X**   |         | Path of the file to write to. Its parent directory is created if needed |
+| `file_path` | `string` |  **X**   |         | Path of the file to write to. Its parent directory is created if needed (local path only) |
 | `inline`    | `int`    |          | `4`     | Level at which the dumper switches to inline YAML (see `Yaml::dump()`) |
 
 Examples
@@ -44,4 +44,6 @@ Notes
 
 * This task is not blocking: when receiving several inputs, each one overwrites the file. Aggregate the data first
   (e.g. with [AggregateIterableTask](aggregate_iterable_task.md)) if needed.
+* The parent directory is created only for a local path (no scheme, or `file://`): a stream wrapper path
+  (`compress.zlib://`, `s3://`, `php://stdout`...) is written as is, its directory is never created.
 * When the file cannot be written (e.g. not writable directory), the task fails with a `RuntimeException`.

docs/reference/transformers/rules_transformer.md

@@ -39,7 +39,7 @@ Each rule of `rules_set` has the following options:
 
 | Code           | Type           | Required | Default | Description                                                                                                                           |
 |----------------|----------------|:--------:|---------|---------------------------------------------------------------------------------------------------------------------------------------|
-| `condition`    | `string\|ParsedExpression\|null` |          | `null`  | Expression (or an already parsed `Symfony\Component\ExpressionLanguage\ParsedExpression`); the rule matches if it is truthy |
+| `condition`    | `string\|ParsedExpression\|null` |          | `null`  | Expression (or an already parsed `Symfony\Component\ExpressionLanguage\ParsedExpression`); the rule matches if its result is truthy (cast to `bool`: `1` of `preg_match()` matches, `0`, `''`, `null` do not) |
 | `default`      | `bool`         |          | `false` | Mark the rule as the default one (always matches). It cannot have a `condition`, no conditional rule may follow it, only one allowed  |
 | `set_null`     | `bool`         |          | `false` | If `true`, return `null` (takes precedence on `constant` and `transformers`)                                                          |
 | `constant`     | `any`          |          | `null`  | If not `null`, return this value (takes precedence on `transformers`)                                                                 |
@@ -88,4 +88,6 @@ Notes
 -----
 
 Conditions are parsed with the bundle's `cleverage_process.expression_language` service, which also exposes the PHP
-`preg_match` function.
+`preg_match` function. The result of a condition does not need to be a boolean: it is cast to `bool`, so a
+`preg_match('/^foo/', value)` (`1` or `0`), a `matches` expression or a single variable (`value`) can be used as is.
+Beware that PHP truthiness applies: the strings `"0"` and `""`, `0`, `[]` and `null` do not match.

src/Exception/InvalidProcessConfigurationException.php

@@ -44,4 +44,19 @@ public static function createEntryPointHasAncestors(
             "The entry-point '{$taskConfig->getCode()}' cannot have an ancestor (from process: {$processConfiguration->getCode()})"
         );
     }
+
+    /**
+     * A task code referenced by the "outputs" or "error_outputs" of a task, or by the "entry_point" or "end_point" of
+     * the process, is not a task of the process.
+     */
+    public static function createUnknownTaskReference(
+        string $processCode,
+        string $key,
+        string $unknownTaskCode,
+        ?string $taskCode = null,
+    ): self {
+        $holder = null === $taskCode ? "the '{$key}'" : "the '{$key}' of the task '{$taskCode}'";
+
+        return new self("Unknown task '{$unknownTaskCode}' in {$holder} (from process: {$processCode})");
+    }
 }

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;