#174 - Reset the process execution state between two messages of a worker

CHANGELOG.md

@@ -12,6 +12,7 @@ Latest
 * [#150](https://github.com/cleverage/ui-process-bundle/issues/150) Require `symfony/monolog-bundle` and `monolog/monolog`, used directly by the bundle (log handlers, `monolog` configuration prepended, `monolog.logger` tags) but only installed through `cleverage/process-bundle`, which may stop requiring MonologBundle (cleverage/process-bundle#269).
 
 ## Fixes
+* [#174](https://github.com/cleverage/ui-process-bundle/issues/174) Long-running workers (Messenger, scheduler): `ProcessExecutionManager` implements `ResetInterface` (`kernel.reset` tag), its current and last process executions are reset between two messages; `DoctrineProcessHandler` drops the log records not written at the end of the previous execution on `reset()` (`kernel.reset` tag). `ProcessEventSubscriber` forgets the ended process execution (and closes its log file) even when it cannot be saved: the next process executed by the worker was attached to it. Add tests.
 * [#173](https://github.com/cleverage/ui-process-bundle/issues/173) Fix the hard-coded URLs: the page displayed after the login is generated from the dashboard route (was `/process`: wrong when the routes are imported with a `prefix`, or the application is served from a sub-directory), and the logo URL is built by the Asset component (base path, `framework.assets.base_urls`, version strategy; absolute URLs kept as is; was `/<logo_path>`). New `ProcessDashboardController` constructor argument (`Symfony\Component\Asset\Packages`, autowired). Require `symfony/asset`, used directly. Add tests.
 * [#171](https://github.com/cleverage/ui-process-bundle/issues/171) Configuration: `config:dump-reference clever_age_ui_process` and `debug:config clever_age_ui_process` no longer fail (`CleverAgeUiProcessExtension::getConfiguration()` gives the kernel environment to `Configuration`, so the PHP config builder is generated too); the options are documented with `info()`. The `logs.database_level`, `logs.file_level` and `logs.report_increment_level` options are validated when the container is compiled, an invalid level no longer makes every process execution fail at runtime (environment variables are still allowed, checked at runtime). Update documentation, add tests.
 * [#169](https://github.com/cleverage/ui-process-bundle/issues/169) Security: the configuration added by the bundle to the `security` extension (`process_user_provider` user provider, provider, login form, logout and API token authenticator of the `main` firewall) overrides the application one, it can now be disabled with the new `security.configure_firewall` option (default `true`, same behaviour as before; `false`: no `security` configuration is added, the application configures the provider and the firewall, e.g. a firewall dedicated to the UI). The `monolog`, `doctrine_migrations` and `security` configurations are skipped when the bundle is not enabled, instead of a `LogicException`. The documentation no longer says that these configurations are prepended (only the Messenger one is). Update documentation, add tests.

config/services/manager.yaml

@@ -4,6 +4,8 @@ services:
         public: false
         arguments:
             - '@cleverage_ui_process.repository.process_execution'
+        tags:
+            - { name: kernel.reset, method: reset }
 
     cleverage_ui_process.manager.process_configuration:
         class: CleverAge\UiProcessBundle\Manager\ProcessConfigurationsManager

config/services/monolog_handler.yaml

@@ -5,6 +5,8 @@ services:
         calls:
             - [ setEntityManager, [ '@doctrine.orm.entity_manager' ] ]
             - [ setProcessExecutionManager, [ '@cleverage_ui_process.manager.process_execution' ] ]
+        tags:
+            - { name: kernel.reset, method: reset }
     CleverAge\UiProcessBundle\Monolog\Handler\DoctrineProcessHandler:
         alias: cleverage_ui_process.monolog_handler.doctrine_process
 

src/EventSubscriber/ProcessEventSubscriber.php

@@ -55,9 +55,7 @@ public function success(ProcessEvent $event): void
         if ($event->getProcessCode() === $this->processExecutionManager->getCurrentProcessExecution()?->getCode()) {
             $this->processExecutionManager->getCurrentProcessExecution()->setStatus(ProcessExecutionStatus::Finish);
             $this->processExecutionManager->getCurrentProcessExecution()->end();
-            $this->processExecutionManager->save()->unsetProcessExecution($event->getProcessCode());
-            $this->processHandler->close();
-            $this->dispatchEnded($event);
+            $this->endProcessExecution($event);
         }
     }
 
@@ -68,10 +66,23 @@ public function fail(ProcessEvent $event): void
                 $this->isStopped($event->getProcessError()) ? ProcessExecutionStatus::Stopped : ProcessExecutionStatus::Failed
             );
             $this->processExecutionManager->getCurrentProcessExecution()->end();
-            $this->processExecutionManager->save()->unsetProcessExecution($event->getProcessCode());
+            $this->endProcessExecution($event);
+        }
+    }
+
+    /**
+     * Save the ended process execution, then forget it even if the save fails (it would stay the current one, and the
+     * next process would be attached to it).
+     */
+    private function endProcessExecution(ProcessEvent $event): void
+    {
+        try {
+            $this->processExecutionManager->save();
+        } finally {
+            $this->processExecutionManager->unsetProcessExecution($event->getProcessCode());
             $this->processHandler->close();
-            $this->dispatchEnded($event);
         }
+        $this->dispatchEnded($event);
     }
 
     /**

src/Manager/ProcessExecutionManager.php

@@ -15,8 +15,9 @@
 
 use CleverAge\UiProcessBundle\Entity\ProcessExecution;
 use CleverAge\UiProcessBundle\Repository\ProcessExecutionRepository;
+use Symfony\Contracts\Service\ResetInterface;
 
-class ProcessExecutionManager
+class ProcessExecutionManager implements ResetInterface
 {
     private ?ProcessExecution $currentProcessExecution = null;
 
@@ -60,6 +61,16 @@ public function getLastProcessExecution(): ?ProcessExecution
         return $this->lastProcessExecution;
     }
 
+    /**
+     * Forget the current and last process executions, e.g. between two messages of a long-running worker: the end of
+     * an execution may have failed before unsetProcessExecution(), the next process would be attached to it.
+     */
+    public function reset(): void
+    {
+        $this->currentProcessExecution = null;
+        $this->lastProcessExecution = null;
+    }
+
     public function unsetProcessExecution(string $processCode): self
     {
         if ($this->currentProcessExecution?->code === $processCode) {

src/Monolog/Handler/DoctrineProcessHandler.php

@@ -85,6 +85,17 @@ public function flush(): void
         $this->records = new ArrayCollection();
     }
 
+    /**
+     * Drop the records not written at the end of the previous process execution (failed flush), e.g. between two
+     * messages of a long-running worker: they would be attached to the next one.
+     */
+    #[\Override]
+    public function reset(): void
+    {
+        parent::reset();
+        $this->records = new ArrayCollection();
+    }
+
     protected function write(LogRecord $record): void
     {
         if (!$this->enabled) {

tests/DependencyInjection/CleverAgeUiProcessExtensionTest.php

@@ -44,6 +44,9 @@ public function testLoadWithDefaultConfiguration(): void
 
         // Services of every config/services/*.yaml file are loaded
         self::assertTrue($container->hasDefinition('cleverage_ui_process.manager.process_execution'));
+        // Stateful services, reset between two messages of a worker
+        self::assertSame([['method' => 'reset']], $container->getDefinition('cleverage_ui_process.manager.process_execution')->getTag('kernel.reset'));
+        self::assertSame([['method' => 'reset']], $container->getDefinition('cleverage_ui_process.monolog_handler.doctrine_process')->getTag('kernel.reset'));
         self::assertTrue($container->hasParameter('upload_directory'));
 
         self::assertSame(

tests/EventSubscriber/ProcessEventSubscriberTest.php

@@ -33,6 +33,7 @@
 use PHPUnit\Framework\Attributes\UsesClass;
 use PHPUnit\Framework\TestCase;
 use Symfony\Component\EventDispatcher\EventDispatcher;
+use Symfony\Contracts\EventDispatcher\EventDispatcherInterface;
 
 #[CoversClass(ProcessEventSubscriber::class)]
 #[UsesClass(DoctrineProcessHandler::class)]
@@ -140,6 +141,38 @@ public function testProcessEnd(string $method, ProcessExecutionStatus $expectedS
         self::assertFalse($processHandler->hasFilename());
     }
 
+    /**
+     * The process execution is forgotten even if it cannot be saved: the next process is not attached to it.
+     *
+     * @param 'success'|'fail' $method
+     */
+    #[DataProvider('provideEnds')]
+    public function testProcessEndWhenTheSaveFails(string $method, ProcessExecutionStatus $expectedStatus): void
+    {
+        $repository = $this->createStub(ProcessExecutionRepository::class);
+        $repository->method('getManaged')->willReturnArgument(0);
+        $repository->method('save')->willThrowException(new \RuntimeException('Database error'));
+        $processExecutionManager = new ProcessExecutionManager($repository);
+        $processExecutionManager->setCurrentProcessExecution(new ProcessExecution('test.process', 'test.log'));
+        $processHandler = new ProcessHandler('/var/log/process', $processExecutionManager);
+        $processHandler->setFilename('test.process/test.log');
+        $eventDispatcher = $this->createMock(EventDispatcherInterface::class);
+        $eventDispatcher->expects(self::never())->method('dispatch');
+        $doctrineProcessHandler = new DoctrineProcessHandler();
+        $doctrineProcessHandler->disable();
+        $subscriber = new ProcessEventSubscriber($processHandler, $doctrineProcessHandler, $processExecutionManager, $eventDispatcher);
+
+        try {
+            $subscriber->{$method}(new ProcessEvent('test.process'));
+            self::fail('The save error should be thrown');
+        } catch (\RuntimeException $exception) {
+            self::assertSame('Database error', $exception->getMessage());
+        }
+
+        self::assertNull($processExecutionManager->getCurrentProcessExecution());
+        self::assertFalse($processHandler->hasFilename());
+    }
+
     /**
      * @return iterable<string, array{\Throwable}>
      */

tests/Manager/ProcessExecutionManagerTest.php

@@ -50,6 +50,24 @@ public function testTheFirstProcessExecutionIsKept(): void
         self::assertSame($first, $manager->getCurrentProcessExecution());
     }
 
+    public function testReset(): void
+    {
+        $processExecution = new ProcessExecution('test.process', 'test.log');
+        $manager = new ProcessExecutionManager($this->createRepositoryStub());
+        $manager->setCurrentProcessExecution(new ProcessExecution('ended.process', 'ended.log'));
+        $manager->unsetProcessExecution('ended.process');
+        $manager->setCurrentProcessExecution($processExecution);
+
+        $manager->reset();
+
+        self::assertNull($manager->getCurrentProcessExecution());
+        self::assertNull($manager->getLastProcessExecution());
+        // The next process execution is the current one
+        $next = new ProcessExecution('next.process', 'next.log');
+        $manager->setCurrentProcessExecution($next);
+        self::assertSame($next, $manager->getCurrentProcessExecution());
+    }
+
     public function testUnsetProcessExecutionOnlyForItsCode(): void
     {
         $processExecution = new ProcessExecution('test.process', 'test.log');

tests/Monolog/Handler/DoctrineProcessHandlerTest.php

@@ -154,6 +154,19 @@ public function testWithoutEntityManagerNorProcessExecutionManager(): void
         $handler->flush();
     }
 
+    public function testResetDropsTheRecordsNotWritten(): void
+    {
+        $entityManager = $this->createMock(EntityManagerInterface::class);
+        $entityManager->method('isOpen')->willReturn(true);
+        $entityManager->expects(self::never())->method('persist');
+
+        $handler = $this->createHandler($entityManager, new ProcessExecution('test.process', 'test.log'));
+        $handler->handle($this->createRecord(Level::Info, 'not written'));
+        $handler->reset();
+        $handler->flush();
+        $handler->disable(); // no flush on destruction
+    }
+
     private function createHandler(EntityManagerInterface $entityManager, ?ProcessExecution $processExecution): DoctrineProcessHandler
     {
         $processExecutionManager = new ProcessExecutionManager($this->createRepositoryStub());