#266 - Allow to stop a running process with stop checkers

CHANGELOG.md

@@ -1,6 +1,9 @@
 Latest
 ------
 
+## Changes
+* [#266](https://github.com/cleverage/process-bundle/issues/266) Allow to stop a running process from the outside: services implementing the new `ProcessStopCheckerInterface` (autoconfigured, tag `cleverage_process.stop_checker`, new optional `$stopCheckers` argument of `ProcessManager`) are asked before each task execution; when one of them returns `true`, the task is not executed and the process fails with a `ProcessFailedException` whose previous exception is the new `ProcessStoppedException`. Add documentation and tests.
+
 v5.2
 -----
 

config/services/manager.yaml

@@ -9,6 +9,7 @@ services:
             - '@cleverage_process.registry.process_configuration'
             - '@cleverage_process.context.contextual_option_resolver'
             - '@event_dispatcher'
+            - !tagged_iterator cleverage_process.stop_checker
 
     cleverage_process.context.contextual_option_resolver:
         class: CleverAge\ProcessBundle\Context\ContextualOptionResolver

docs/04-advanced_workflow.md

@@ -241,6 +241,60 @@ of a process: it dispatches a `CleverAge\ProcessBundle\Event\EventDispatcherTask
 `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
+tasks are not finalized.
+
+```php
+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](reference/tasks/process_launcher_task.md)) 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](reference/tasks/process_executor_task.md) 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.
+
 ## Parallelization
 
 PHP executes a process in a single thread. To use several CPU cores, the

src/DependencyInjection/CleverAgeProcessExtension.php

@@ -13,6 +13,7 @@
 
 namespace CleverAge\ProcessBundle\DependencyInjection;
 
+use CleverAge\ProcessBundle\Model\ProcessStopCheckerInterface;
 use CleverAge\ProcessBundle\Transformer\GenericTransformer;
 use Symfony\Component\Config\FileLocator;
 use Symfony\Component\DependencyInjection\ContainerBuilder;
@@ -34,6 +35,9 @@ public function load(array $configs, ContainerBuilder $container): void
         // Get the path of the service folder wherever the bundle is installed
         $this->findServices($container, __DIR__.'/../../config/services');
 
+        $container->registerForAutoconfiguration(ProcessStopCheckerInterface::class)
+            ->addTag('cleverage_process.stop_checker');
+
         $configuration = new Configuration();
         $config = $this->processConfiguration($configuration, $configs);
 

src/Exception/ProcessStoppedException.php

@@ -0,0 +1,26 @@
+<?php
+
+declare(strict_types=1);
+
+/*
+ * This file is part of the CleverAge/ProcessBundle package.
+ *
+ * Copyright (c) Clever-Age
+ *
+ * For the full copyright and license information, please view the LICENSE
+ * file that was distributed with this source code.
+ */
+
+namespace CleverAge\ProcessBundle\Exception;
+
+/**
+ * Set on the task state when a process is stopped by a ProcessStopCheckerInterface: the process fails with a
+ * ProcessFailedException, whose previous exception is this one.
+ */
+class ProcessStoppedException extends \RuntimeException implements ProcessExceptionInterface
+{
+    public static function create(string $processCode, string $taskCode): self
+    {
+        return new self("Process {$processCode} has been stopped before task {$taskCode}.");
+    }
+}

src/Manager/ProcessManager.php

@@ -19,6 +19,7 @@
 use CleverAge\ProcessBundle\Event\ProcessEvent;
 use CleverAge\ProcessBundle\Exception\InvalidProcessConfigurationException;
 use CleverAge\ProcessBundle\Exception\ProcessFailedException;
+use CleverAge\ProcessBundle\Exception\ProcessStoppedException;
 use CleverAge\ProcessBundle\Logger\ProcessLogger;
 use CleverAge\ProcessBundle\Logger\TaskLogger;
 use CleverAge\ProcessBundle\Model\BlockingTaskInterface;
@@ -28,6 +29,7 @@
 use CleverAge\ProcessBundle\Model\IterableTaskInterface;
 use CleverAge\ProcessBundle\Model\ProcessHistory;
 use CleverAge\ProcessBundle\Model\ProcessState;
+use CleverAge\ProcessBundle\Model\ProcessStopCheckerInterface;
 use CleverAge\ProcessBundle\Model\TaskInterface;
 use CleverAge\ProcessBundle\Registry\ProcessConfigurationRegistry;
 use Symfony\Component\DependencyInjection\ContainerInterface;
@@ -58,13 +60,17 @@ class ProcessManager
 
     protected ?TaskConfiguration $taskConfiguration = null;
 
+    /**
+     * @param iterable<ProcessStopCheckerInterface> $stopCheckers
+     */
     public function __construct(
         protected ContainerInterface $container,
         protected ProcessLogger $processLogger,
         protected TaskLogger $taskLogger,
         protected ProcessConfigurationRegistry $processConfigurationRegistry,
         protected ContextualOptionResolver $contextualOptionResolver,
         protected EventDispatcherInterface $eventDispatcher,
+        protected iterable $stopCheckers = [],
     ) {
     }
 
@@ -377,6 +383,14 @@ protected function processExecution(TaskConfiguration $taskConfiguration, int $e
         }
         $state = $taskConfiguration->getState();
 
+        if ($this->isStopRequested($state)) {
+            $state->reset(self::EXECUTE_PROCESS !== $executionFlag);
+            $this->processLogger->info("Process stop requested, the task {$taskConfiguration->getCode()} is not executed");
+            $state->stop(ProcessStoppedException::create($state->getProcessConfiguration()->getCode(), $taskConfiguration->getCode()));
+
+            return;
+        }
+
         try {
             if (self::EXECUTE_PROCESS === $executionFlag) {
                 $state->reset(false);
@@ -432,6 +446,17 @@ protected function processExecution(TaskConfiguration $taskConfiguration, int $e
         }
     }
 
+    protected function isStopRequested(ProcessState $state): bool
+    {
+        foreach ($this->stopCheckers as $stopChecker) {
+            if ($stopChecker->shouldStop($state)) {
+                return true;
+            }
+        }
+
+        return false;
+    }
+
     /**
      * Browse all children for FlushableTask until a BlockingTask is found.
      */

src/Model/ProcessStopCheckerInterface.php

@@ -0,0 +1,33 @@
+<?php
+
+declare(strict_types=1);
+
+/*
+ * This file is part of the CleverAge/ProcessBundle package.
+ *
+ * Copyright (c) Clever-Age
+ *
+ * For the full copyright and license information, please view the LICENSE
+ * file that was distributed with this source code.
+ */
+
+namespace CleverAge\ProcessBundle\Model;
+
+/**
+ * Allows to stop a running process from the outside (e.g. on a request made from a UI).
+ *
+ * The process manager asks every checker before each task execution (once per item for the iterable tasks): an
+ * implementation must be fast, e.g. check an external storage at most every few seconds. When a checker returns true,
+ * the process is stopped before executing the task, as a task error with the "stop" error strategy, with a
+ * ProcessStoppedException.
+ *
+ * Services implementing this interface are autoconfigured with the "cleverage_process.stop_checker" tag.
+ */
+interface ProcessStopCheckerInterface
+{
+    /**
+     * The state is the one of the task about to be executed, which may belong to a sub-process
+     * (ProcessExecutorTask): keep returning true until the process has failed.
+     */
+    public function shouldStop(ProcessState $state): bool;
+}

tests/DependencyInjection/CleverAgeProcessExtensionTest.php

@@ -0,0 +1,42 @@
+<?php
+
+declare(strict_types=1);
+
+/*
+ * This file is part of the CleverAge/ProcessBundle package.
+ *
+ * Copyright (c) Clever-Age
+ *
+ * For the full copyright and license information, please view the LICENSE
+ * file that was distributed with this source code.
+ */
+
+namespace CleverAge\ProcessBundle\Tests\DependencyInjection;
+
+use CleverAge\ProcessBundle\DependencyInjection\CleverAgeProcessExtension;
+use CleverAge\ProcessBundle\DependencyInjection\Configuration;
+use CleverAge\ProcessBundle\Model\ProcessStopCheckerInterface;
+use PHPUnit\Framework\Attributes\CoversClass;
+use PHPUnit\Framework\Attributes\UsesClass;
+use PHPUnit\Framework\TestCase;
+use Symfony\Component\DependencyInjection\Argument\TaggedIteratorArgument;
+use Symfony\Component\DependencyInjection\ContainerBuilder;
+
+#[CoversClass(CleverAgeProcessExtension::class)]
+#[UsesClass(Configuration::class)]
+class CleverAgeProcessExtensionTest extends TestCase
+{
+    public function testStopCheckersAreInjectedInTheProcessManager(): void
+    {
+        $container = new ContainerBuilder();
+        (new CleverAgeProcessExtension())->load([], $container);
+
+        $autoconfigured = $container->getAutoconfiguredInstanceof();
+        self::assertArrayHasKey(ProcessStopCheckerInterface::class, $autoconfigured);
+        self::assertSame(['cleverage_process.stop_checker' => [[]]], $autoconfigured[ProcessStopCheckerInterface::class]->getTags());
+
+        $argument = $container->getDefinition('cleverage_process.manager.process')->getArgument(6);
+        self::assertInstanceOf(TaggedIteratorArgument::class, $argument);
+        self::assertSame('cleverage_process.stop_checker', $argument->getTag());
+    }
+}