04-advanced_workflow.md

Advanced Workflow

Process execution flow

When a process is executed (with the cleverage:process:execute command or with CleverAge\ProcessBundle\Manager\ProcessManager::execute($processCode, $input, $context)), the process manager:

1. dispatches the cleverage_process.start event
2. checks the process: circular dependencies are forbidden, and the entry_point / end_point tasks must be part of the main branch, the group of connected tasks that is actually executed (see orphan tasks)
3. initializes every task, in the order they are configured: the service is fetched from the container and initialize is called on initializable tasks (for configurable tasks, this is where options are validated). An exception thrown by initialize is logged as critical and flags the task as stopped, but does not abort the process: it only fails when this task is first executed. This is deprecated since 5.1 (an E_USER_DEPRECATED error is triggered): in v6.0, the process will fail at this step, before any task is executed
4. gives the process input to the entry_point task, if one is defined (otherwise the input is ignored and a warning is logged)
5. resolves the tasks of the main branch (see below)
6. finalizes every task, in the order they are configured (see finalizable tasks)
7. returns the last output of the end_point task passed on to its next tasks (or null if there is none: an output that is ignored, when the task is skipped, filters its input or fails with the skip strategy, does not replace it) and dispatches the cleverage_process.end event (ProcessEvent::isProcessStopped() tells whether the process was stopped on purpose by a StopTask)

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

The exceptions thrown by the bundle implement CleverAge\ProcessBundle\Exception\ProcessExceptionInterface (since 5.4 for the UnexpectedValueException, RuntimeException, InvalidArgumentException and LogicException of the same namespace, which extend the SPL exceptions thrown before), so that they can all be caught at once. Not concerned: the exceptions of the Symfony components (e.g. InvalidOptionsException for an invalid task option) and the ones of your own tasks and transformers. The reference pages name these exceptions without their namespace (e.g. UnexpectedValueException). In a chain of transformers, the exception of a transformer is wrapped in a TransformerException (see TransformerTrait).

Executing a process from PHP

The process manager service (cleverage_process.manager.process, private) can be autowired by its interface, CleverAge\ProcessBundle\Manager\ProcessManagerInterface (or by its class, ProcessManager), or injected explicitly. Type your services on the interface: the class will be final in v6.0 (the services of the bundle will then be typed on the interfaces, so that they can be decorated). The registries have their interface too: ProcessConfigurationRegistryInterface, TransformerRegistryInterface and TaskRegistryInterface (namespace CleverAge\ProcessBundle\Registry).

# config/services.yaml
services:
    App\Service\ProductImporter:
        arguments:
            $processManager: '@cleverage_process.manager.process' # not needed with autowiring

$result = $this->processManager->execute('app.import_file', '/tmp/products.csv', ['dry_run' => true]);

Task resolution & blocking

A task is resolved when it and all its ancestors are over. Tasks are resolved starting from the end of the configuration, each task first resolving its parents:

• a root task (a task without parent, outside of an error branch) is executed; its output is immediately passed to each of its outputs, which are executed in turn, and so on (depth-first). An iterable task is executed again as long as its next method returns true, each item going through the whole following chain before the next one is produced
• a blocking task receives all the inputs of its parents but does not pass anything to its outputs; once all its parents are resolved, proceed is called and its output starts a new flow to its outputs. If the blocking task never received any input, proceed is not called
• when a task is resolved (and when an iterable task finishes its iterations), the task itself and the flushable tasks found in its descendants (until a blocking task) are flushed. As each resolved ancestor triggers this, a flushable task can be flushed several times: flush must be idempotent

The following process reads a CSV file line by line, transforms each line and writes it into another CSV file: the reader is iterable, the transformer is executed once for each line, and the writer is blocking: it writes each line and outputs the file path only once, at the end, to the log task.

clever_age_process:
    configurations:
        app.csv_copy:
            tasks:
                read:
                    task: csv_reader
                    options:
                        file_path: '%kernel.project_dir%/var/data/input.csv'
                    outputs: [transform]
                transform:
                    task: transformer
                    options:
                        transformers:
                            mapping:
                                mapping:
                                    id: { code: '[id]' }
                                    name: { code: '[name]' }
                    outputs: [write]
                write:
                    task: csv_writer
                    options:
                        file_path: '%kernel.project_dir%/var/data/output.csv'
                    outputs: [log]
                log:
                    task: debug

Orphan tasks

Only one group of connected tasks, the main branch, is executed. It is the group containing the entry_point task, or the end_point task if there is no entry point, or else the first configured task. Tasks that are not connected to it (not referenced in any outputs or error_outputs of the main branch) are never executed: a warning Task '<code>' is unreachable is logged for each of them. They are still initialized and finalized, like every task of the process (so a misconfigured orphan task still logs its initialization error). The cleverage:process:help command only displays the main branch.

Errors and skips

An error happens when a task throws an exception, or flags one with ProcessState::setException(). The error is logged on the cleverage_process_task channel with the log_level of the task (critical by default), then the error_strategy of the task is applied (it defaults to the global default_error_strategy, itself stop by default):

• skip: the current output is dropped, and the process continues with the next input (e.g. the next line of a CSV file)
• stop: the whole process stops and fails (a ProcessFailedException is thrown by the process manager, see process execution flow)

Before applying the strategy, the task input is sent to the tasks listed in error_outputs (unless the task already set a specific value with ProcessState::setErrorOutput()). This input is null when the task has none: a root task (except the entry_point, which receives the process input), and a failure of proceed (blocking tasks) or flush (flushable tasks), which are called without input. A task being flushed several times, a flush failing with the skip strategy (e.g. keeping the items it could not write) sends null to its error branch at each flush. Those tasks form an error branch: they are only executed when an error output is set, and they are never considered as root tasks. A task can also send a value to its error branch without failing, by calling setErrorOutput() (e.g. FilterTask sends rejected items to its error outputs). With the stop strategy, the error branch is executed before the process stops, and the process fails even if the error branch stops the flow (e.g. with a StopTask).

transform:
    task: transformer
    error_strategy: skip          # Continue with the next item on error
    log_level: warning            # Log errors as warning instead of critical
    options:
        transformers:
            mapping:
                mapping:
                    price: { code: '[price]', transformers: { cast: { type: float } } }
    outputs: [write]
    error_outputs: [write_errors] # Receives the input that failed
write_errors:
    task: json_stream_writer
    options:
        file_path: '%kernel.project_dir%/var/data/errors_{date_time}.json-stream'

A task may also control the flow without any error:

• ProcessState::setSkipped(true): the output of this execution is not sent to the next tasks (used by filters, by iterable tasks on empty items, by batch tasks while filling a batch...)
• ProcessState::setStopped(true) / ProcessState::stop(): without exception, only the current flow stops: the stop signal goes back up to the root task, which stops iterating. The rest of the resolution still happens: other root tasks are executed, blocking tasks that already received inputs are proceeded, flushable tasks are flushed and every task is finalized. Their outputs go on through the tasks of the stopped flow, except through an iterable task whose iteration has been interrupted by the stop: it stays stopped (it would resume its interrupted iteration), and stops the flows reaching it. It is not a failure for the process manager: the cleverage_process.end event is dispatched (not cleverage_process.fail) and the command exits with code 0. The StopTask also flags the process history as stopped: the process ends as stopped on purpose, not as a failure

The legacy errors key is a deprecated alias of error_outputs (defining both throws an exception).

Wrapping execution in subprocesses

A process can execute another process with the ProcessExecutorTask: for each input, the sub-process is executed in the same PHP process, with this input given to its entry_point, and the last output of its end_point becomes the output of the task (null if the sub-process has no end_point). Context values are not inherited: pass them explicitly with the context option (placeholders like {{ key }} can be used to forward the parent context).

clever_age_process:
    configurations:
        app.import_all:
            tasks:
                list_files:
                    task: folder_browser
                    options:
                        folder_path: '%kernel.project_dir%/var/import'
                        name_pattern: '*.csv'
                    outputs: [import_file]
                import_file:
                    task: process_executor
                    options:
                        process: app.import_file
                        context:
                            dry_run: '{{ dry_run }}' # Always pass -c dry_run:<value>, see below

        app.import_file:
            public: false
            entry_point: read
            tasks:
                read:
                    task: input_csv_reader
                    outputs: [save]
                save:
                    task: app_save_product # code of an application task, see #[AsTask]

If the parent process is executed without a dry_run context value, the placeholder is not replaced and the sub-process receives the literal string '{{ dry_run }}', which is truthy: always provide the forwarded context values when executing the parent process.

This is a good way to split a big workflow into small, testable processes. Private processes (public: false) are hidden from cleverage:process:list (unless --all is used) but can still be executed.

Executing a process while it is already running, by itself or through its sub-processes, is deprecated and will throw an exception in 6.0 (see the ProcessExecutorTask).

Events

Events are dispatched around process execution (see CleverAge\ProcessBundle\Event\ProcessEvent, which gives access to the process code, input, context and, depending on the event, output or error):

Event name	Constant	Dispatched
cleverage_process.start	ProcessEvent::EVENT_PROCESS_STARTED	on process start
cleverage_process.end	ProcessEvent::EVENT_PROCESS_ENDED	on successful process end (with the process output), including a process stopped on purpose (isProcessStopped())
cleverage_process.fail	ProcessEvent::EVENT_PROCESS_FAILED	on failed process end (with the associated error)

Those events are also dispatched for sub-processes executed with the ProcessExecutorTask.

An exception thrown by a listener of cleverage_process.fail is logged as an error (A listener of the process failure has failed) and does not replace the error of the process, which is rethrown (the following listeners of the event are not called).

Another event is dispatched once by the cleverage:process:execute command, before executing any process: CleverAge\ProcessBundle\Event\ConsoleProcessEvent. It is dispatched without a specific name, so listeners must use the class name as event name. It gives access to the console input/output objects, and to the process input and context.

namespace App\EventListener;

use CleverAge\ProcessBundle\Event\ProcessEvent;
use Psr\Log\LoggerInterface;
use Symfony\Component\EventDispatcher\Attribute\AsEventListener;

#[AsEventListener(event: ProcessEvent::EVENT_PROCESS_FAILED)]
class ProcessFailureListener
{
    public function __construct(private LoggerInterface $logger)
    {
    }

    public function __invoke(ProcessEvent $event): void
    {
        $this->logger->alert("Process {$event->getProcessCode()} failed", [
            'error' => $event->getProcessError()?->getMessage(),
        ]);
    }
}

You can also use the EventDispatcherTask to trigger an event in the middle of a process: it dispatches a CleverAge\ProcessBundle\Event\EventDispatcherTaskEvent, giving access to the current ProcessState. The event is dispatched under the event_name option or, when it is not set, under its class name (CleverAge\ProcessBundle\Event\EventDispatcherTaskEvent).

Stopping a running process

A running process can be stopped from the outside (e.g. on a request made from a UI) by a service implementing CleverAge\ProcessBundle\Model\ProcessStopCheckerInterface. Such services are autoconfigured (tag cleverage_process.stop_checker) and injected into the process manager, which calls their shouldStop() method before each task execution: once per item for the iterable tasks and their children, and before proceeding or flushing a task.

When a checker returns true, the task is not executed and the process is stopped as if the task had failed with the 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 initialized tasks are finalized.

namespace App\Process;

use CleverAge\ProcessBundle\Model\ProcessState;
use CleverAge\ProcessBundle\Model\ProcessStopCheckerInterface;

class FileStopChecker implements ProcessStopCheckerInterface
{
    private float $lastCheck = 0;

    private bool $stop = false;

    public function __construct(private string $stopDirectory)
    {
    }

    public function shouldStop(ProcessState $state): bool
    {
        // Called very often: check at most every 2 seconds
        if (!$this->stop && microtime(true) - $this->lastCheck >= 2) {
            $this->lastCheck = microtime(true);
            $this->stop = file_exists($this->stopDirectory.'/'.$state->getProcessConfiguration()->getCode());
        }

        return $this->stop;
    }
}

Keep in mind that:

• shouldStop() is called very often: it must be fast, e.g. query an external storage at most every few seconds.
• The process stops between two task executions: a task blocked in a single long call (SQL query, HTTP request, sub-process launched by the ProcessLauncherTask) stops only when the call returns. Sub-processes launched by the ProcessLauncherTask are separate PHP processes, which are not stopped.
• Sub-processes executed by the ProcessExecutorTask run in the same process manager: the checkers receive the states of their tasks. Their failure is a task error of the parent process (handled according to the error_strategy of the ProcessExecutorTask): keep returning true until the parent process has failed. To recognize a stopped process, look for a ProcessStoppedException in the chain of the previous exceptions of the process error.

SIGTERM and SIGINT

When a console command executing a process (cleverage:process:execute, messenger:consume...) receives SIGTERM (docker stop, Kubernetes, systemd, supervisord...) or SIGINT (Ctrl+C) during the process, the bundle stops the process before its next task, as a stop checker, instead of exiting right away: the initialized tasks are finalized, the cleverage_process.fail event is dispatched (e.g. the UI records the execution as stopped), a warning is logged on the cleverage_process channel, and cleverage:process:execute exits with 1 without executing the next processes of its arguments. A Messenger worker stops after the message (the message fails with the process). A second signal exits right away, as a signal received when no process is running.

This requires the pcntl extension (the Symfony console application dispatches the signals only with it): without it, the command is killed right away, the tasks are not finalized and the end of the process is never recorded. As for any stop, a task blocked in a single long call stops only when the call returns: give the process enough time to stop before killing it (e.g. docker stop --time, terminationGracePeriodSeconds of Kubernetes, stopwaitsecs of supervisord).

Ctrl+C in a terminal sends SIGINT to the whole foreground process group: the system processes started by the CommandRunnerTask and the ProcessLauncherTask receive it too, and are interrupted right away: the task fails (e.g. a ProcessSignaledException for the CommandRunnerTask), instead of the cooperative stop. A signal sent to the PHP process only (kill -TERM <pid>, docker stop for the main process of a container) does not reach them.

Parallelization

PHP executes a process in a single thread. To use several CPU cores, the ProcessLauncherTask launches a process in a separate system process (bin/console cleverage:process:execute --input-from-stdin ..., in the same environment) for each input it receives:

• at most max_processes sub-processes run at the same time; the task waits for a free slot before launching a new one
• the input is cast to string and sent to the sub-process through STDIN, context values are passed as --context=<key>:<value> options (values are cast to string, so arrays cannot be forwarded this way)
• with json_buffering enabled, the sub-process runs with --output-format=json-stream --output=<file>: the last output 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 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 (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

parallel_import:
    task: process_launcher
    options:
        process: app.import_file
        max_processes: 4

Since sub-processes do not share memory with the parent process, their input must be small and serializable as a string (an identifier, a file path, a chunk produced by the CsvSplitterTask...).

To execute any other system command (not a process), use the CommandRunnerTask: it runs the commandline option synchronously with the Symfony Process component, sends the task input to its STDIN, and outputs what the command wrote on STDOUT.