diff --git a/CHANGELOG.md b/CHANGELOG.md index aca8d079..f9f63be3 100644 --- a/CHANGELOG.md +++ b/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 ----- diff --git a/config/services/manager.yaml b/config/services/manager.yaml index 396d7d9f..8445bb42 100644 --- a/config/services/manager.yaml +++ b/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 diff --git a/docs/04-advanced_workflow.md b/docs/04-advanced_workflow.md index e73ac33a..a30ff54e 100644 --- a/docs/04-advanced_workflow.md +++ b/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 diff --git a/src/DependencyInjection/CleverAgeProcessExtension.php b/src/DependencyInjection/CleverAgeProcessExtension.php index 1c8c5cf1..5283ce8c 100644 --- a/src/DependencyInjection/CleverAgeProcessExtension.php +++ b/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); diff --git a/src/Exception/ProcessStoppedException.php b/src/Exception/ProcessStoppedException.php new file mode 100644 index 00000000..74c26a48 --- /dev/null +++ b/src/Exception/ProcessStoppedException.php @@ -0,0 +1,26 @@ + $stopCheckers + */ public function __construct( protected ContainerInterface $container, protected ProcessLogger $processLogger, @@ -65,6 +70,7 @@ public function __construct( 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. */ diff --git a/src/Model/ProcessStopCheckerInterface.php b/src/Model/ProcessStopCheckerInterface.php new file mode 100644 index 00000000..6a95e315 --- /dev/null +++ b/src/Model/ProcessStopCheckerInterface.php @@ -0,0 +1,33 @@ +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()); + } +} diff --git a/tests/Manager/ProcessManagerStopCheckerTest.php b/tests/Manager/ProcessManagerStopCheckerTest.php new file mode 100644 index 00000000..7c76c3fd --- /dev/null +++ b/tests/Manager/ProcessManagerStopCheckerTest.php @@ -0,0 +1,232 @@ + + */ + private array $recorded = []; + + private ?\Throwable $failedEventError = null; + + public function testWithoutStopCheckerTheProcessRunsEntirely(): void + { + $this->createProcessManager([])->execute('test.process'); + + self::assertSame([1, 2, 3, 4], $this->recorded); + self::assertNull($this->failedEventError); + } + + public function testCheckerNotRequestingAStop(): void + { + $checked = []; + $checker = $this->createChecker(static function (ProcessState $state) use (&$checked): bool { + $checked[] = $state->getTaskConfiguration()->getCode(); + + return false; + }); + + $this->createProcessManager([$checker])->execute('test.process'); + + self::assertSame([1, 2, 3, 4], $this->recorded); + // Before each task execution: once per item for the iterable task and its children + self::assertSame(['iterate', 'record', 'iterate', 'record', 'iterate', 'record', 'iterate', 'record'], $checked); + } + + /** + * The stop is not handled as a task error: the "skip" error strategy of the task does not apply, its error + * outputs are not executed. + */ + public function testStopRequestedDuringTheIteration(): void + { + $checker = $this->createChecker(static fn (ProcessState $state): bool => 'record' === $state->getTaskConfiguration()->getCode() && 3 === $state->getInput()); + $processManager = $this->createProcessManager([$checker]); + + try { + $processManager->execute('test.process'); + self::fail('The process should have failed'); + } catch (ProcessFailedException $exception) { + $stopped = $exception->getPrevious(); + self::assertInstanceOf(ProcessStoppedException::class, $stopped); + self::assertSame('Process test.process has been stopped before task record.', $stopped->getMessage()); + } + + self::assertSame([1, 2], $this->recorded); + self::assertSame($exception, $this->failedEventError); + self::assertTrue($processManager->getProcessHistory()?->isFailed()); + } + + public function testAnyCheckerCanRequestAStop(): void + { + $neverStops = $this->createChecker(static fn (): bool => false); + $alwaysStops = $this->createChecker(static fn (): bool => true); + + $this->expectException(ProcessFailedException::class); + $this->expectExceptionMessage('Process test.process has been stopped before task iterate.'); + + try { + $this->createProcessManager([$neverStops, $alwaysStops])->execute('test.process'); + } finally { + self::assertSame([], $this->recorded); + } + } + + /** + * @param \Closure(ProcessState): bool $shouldStop + */ + private function createChecker(\Closure $shouldStop): ProcessStopCheckerInterface + { + return new class($shouldStop) implements ProcessStopCheckerInterface { + /** + * @param \Closure(ProcessState): bool $shouldStop + */ + public function __construct(private readonly \Closure $shouldStop) + { + } + + public function shouldStop(ProcessState $state): bool + { + return ($this->shouldStop)($state); + } + }; + } + + /** + * @param list $stopCheckers + */ + private function createProcessManager(array $stopCheckers): ProcessManager + { + $recordTask = new class(function (mixed $input): void { + $this->recorded[] = $input; + }) implements TaskInterface { + /** + * @param \Closure(mixed): void $record + */ + public function __construct(private readonly \Closure $record) + { + } + + public function execute(ProcessState $state): void + { + ($this->record)($state->getInput()); + } + }; + + $container = new Container(); + $container->set('test.iterate', new ConstantIterableOutputTask()); + $container->set('test.record', $recordTask); + + $registry = new ProcessConfigurationRegistry( + [ + 'test.process' => [ + 'options' => [], + 'entry_point' => null, + 'end_point' => null, + 'description' => '', + 'help' => '', + 'public' => true, + 'tasks' => [ + 'iterate' => $this->createTaskConfiguration('@test.iterate', ['output' => [1, 2, 3, 4]], ['record']), + 'record' => $this->createTaskConfiguration('@test.record', [], [], 'skip', ['error']), + 'error' => $this->createTaskConfiguration('@test.record'), + ], + ], + ], + 'stop' + ); + + $eventDispatcher = new EventDispatcher(); + $eventDispatcher->addListener(ProcessEvent::EVENT_PROCESS_FAILED, function (ProcessEvent $event): void { + $this->failedEventError = $event->getProcessError(); + }); + + return new ProcessManager( + $container, + new ProcessLogger(new NullLogger()), + new TaskLogger(new NullLogger()), + $registry, + new ContextualOptionResolver(), + $eventDispatcher, + $stopCheckers, + ); + } + + /** + * @param array $options + * @param list $outputs + * @param list $errorOutputs + * + * @return array + */ + private function createTaskConfiguration( + string $service, + array $options = [], + array $outputs = [], + ?string $errorStrategy = null, + array $errorOutputs = [], + ): array { + return [ + 'service' => $service, + 'options' => $options, + 'description' => '', + 'help' => '', + 'outputs' => $outputs, + 'errors' => [], + 'error_outputs' => $errorOutputs, + 'error_strategy' => $errorStrategy, + 'log_level' => null, + ]; + } +}