#230 - Logs in database: store invalid UTF-8 and long messages, do not fail the process when storing the logs fails

CHANGELOG.md

@@ -3,6 +3,7 @@ Latest
 
 ## Fixes
 * [#229](https://github.com/cleverage/ui-process-bundle/issues/229) Login throttling of the `main` firewall (#207, #222): the requests of the HTTP API (`HttpProcessExecutionAuthenticator`) are no longer counted, on every firewall with a login throttling (`LoginThrottlingPass` decorates its limiter, the firewall settings still apply to the login form): failed logins blocked a valid API token, and invalid tokens blocked the login form of the IP address. `login_throttling: false` in the application `main` firewall disables it (it was enabled again by the bundle). With `framework.rate_limiter: false`, an explicit error tells how to fix the configuration instead of "Parent definition "limiter" does not exist". `HttpProcessExecutionAuthenticator::onAuthenticationFailure()` answers the message key of the exception, instead of an empty message when there were too many attempts. Update documentation, add tests.
+* [#230](https://github.com/cleverage/ui-process-bundle/issues/230) `DoctrineProcessHandler`, `LogRecord`: a log record with an invalid UTF-8 sequence in its message or its context (e.g. a Latin-1 file imported), or with a message longer than the column in code points (`LogRecord` truncated it to 512 graphemes: emojis, combining marks), no longer makes the insertion of the logs in database fail: the invalid sequences are replaced (and the NUL characters of the message removed) and the message is truncated to 512 code points. When the insertion of the log records fails anyway, they (up to 500) are dropped and the failure is logged on the `cleverage_process` channel, instead of making the process fail (on the 500th record, even with `error_strategy: skip`) or preventing the status of its execution from being saved at its end (left `started`, the command exiting with `1`); a failure of a flush also writing pending changes of the process tasks (the entity manager is shared) is not caught, as before. Update documentation, add tests.
 
 v3.4
 ------

config/services/monolog_handler.yaml

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

docs/reference/04-process_executions_and_logs.md

@@ -174,7 +174,19 @@ When `store_in_database` is `true` (default), log records are also stored in the
 | `context`   | Log record context.                           |
 | `createdAt` | Date of the log record.                       |
 
-Records are buffered and inserted every 500 records, and at the end (or failure) of the process.
+The lengths are counted in characters (Unicode code points, as MySQL and PostgreSQL do): an emoji or a letter with
+combining marks may count several characters. The invalid UTF-8 sequences of the message and the channel are replaced
+by `?` and their NUL characters removed; a context which cannot be encoded in JSON is stored with its invalid UTF-8 sequences replaced by `U+FFFD` (`�`)
+and its values which cannot be encoded (e.g. a resource) replaced by `null`.
+
+Records are buffered and inserted every 500 records, and at the end (or failure) of the process. When the insertion of
+the log records fails (e.g. a database error), they are dropped and the failure is logged as an error on the
+`cleverage_process` channel (in the [log file](#log-file), not in database): the process goes on, and the status of its
+execution is saved. The entity manager is shared with the process tasks: when changes of their entities are pending
+(e.g. persisted by a task, not flushed yet), they are written with the log records, and a failure is not caught, as any
+error of the tasks. To know it, when entities of the tasks are managed by the entity manager, their changes are computed
+before each insertion of the log records (and again by the flush): a cost proportional to the number of managed
+entities, every 500 log records.
 
 On processes logging a lot, storing logs in database may be slow and memory consuming: raise `database_level`, or
 disable `store_in_database` and rely on the log file (see [troubleshooting](../troubleshooting.md)).

src/Entity/LogRecord.php

@@ -15,7 +15,6 @@
 
 use Doctrine\DBAL\Types\Types;
 use Doctrine\ORM\Mapping as ORM;
-use Symfony\Component\String\UnicodeString;
 
 #[ORM\Entity]
 #[ORM\Table(name: 'log_record')]
@@ -55,13 +54,43 @@ public function __construct(
         #[ORM\JoinColumn(name: 'process_execution_id', referencedColumnName: 'id', nullable: false, onDelete: 'CASCADE')]
         private readonly ProcessExecution $processExecution,
     ) {
-        $this->channel = (string) (new UnicodeString($record->channel))->truncate(64);
+        $this->channel = $this->truncate($record->channel, 64);
         $this->level = $record->level->value;
-        $this->message = (string) (new UnicodeString($record->message))->truncate(512);
-        $this->context = $record->context;
+        $this->message = $this->truncate($record->message, 512);
+        $this->context = $this->encodableContext($record->context);
         $this->createdAt = $record->datetime;
     }
 
+    /**
+     * Invalid UTF-8 sequences replaced (by "?"), NUL characters removed (rejected by PostgreSQL, or the value truncated
+     * there), truncated to the length of the column: a number of characters (code points) in the database, not of
+     * graphemes (an emoji or a letter with combining marks is a grapheme of several code points).
+     */
+    private function truncate(string $value, int $length): string
+    {
+        return mb_substr(str_replace("\0", '', mb_scrub($value, 'UTF-8')), 0, $length, 'UTF-8');
+    }
+
+    /**
+     * The context, stored as JSON: unchanged when it can be encoded, otherwise its JSON representation where the
+     * invalid UTF-8 sequences are replaced (by U+FFFD) and the values which cannot be encoded (e.g. a resource) are
+     * replaced by null: the insertion of the log record, and of the other ones of the batch, would fail.
+     *
+     * @param array<mixed> $context
+     *
+     * @return array<mixed>
+     */
+    private function encodableContext(array $context): array
+    {
+        if (false !== json_encode($context, \JSON_PRESERVE_ZERO_FRACTION)) {
+            return $context;
+        }
+        $json = json_encode($context, \JSON_PRESERVE_ZERO_FRACTION | \JSON_INVALID_UTF8_SUBSTITUTE | \JSON_PARTIAL_OUTPUT_ON_ERROR);
+        $encodable = false === $json ? null : json_decode($json, true);
+
+        return \is_array($encodable) ? $encodable : [];
+    }
+
     /**
      * Not named hasContext(): the PropertyAccessor would use it to read the "context" property (e.g. in EasyAdmin).
      */

src/Monolog/Handler/DoctrineProcessHandler.php

@@ -20,6 +20,7 @@
 use Monolog\Handler\AbstractProcessingHandler;
 use Monolog\Level;
 use Monolog\LogRecord;
+use Psr\Log\LoggerInterface;
 
 /**
  * @internal
@@ -31,6 +32,8 @@ class DoctrineProcessHandler extends AbstractProcessingHandler
     private ?EntityManagerInterface $em = null;
     private ?ProcessExecutionManager $processExecutionManager = null;
     private bool $enabled = true;
+    private ?LoggerInterface $logger = null;
+    private bool $reportingFailure = false;
 
     public function __construct(int|string|Level $level = Level::Debug, bool $bubble = true)
     {
@@ -48,6 +51,14 @@ public function setProcessExecutionManager(ProcessExecutionManager $processExecu
         $this->processExecutionManager = $processExecutionManager;
     }
 
+    /**
+     * Logger reporting the failures to store the logs in database (the log records are dropped, the process goes on).
+     */
+    public function setLogger(?LoggerInterface $logger): void
+    {
+        $this->logger = $logger;
+    }
+
     public function disable(): void
     {
         $this->enabled = false;
@@ -70,27 +81,54 @@ public function flush(): void
             // or saving the process execution, see ProcessExecutionRepository::getManaged())
             return;
         }
+        if ($this->records->isEmpty()) {
+            // Writes the changes pending in the entity manager, shared with the process tasks: their errors are not
+            // caught
+            $this->em?->flush();
+
+            return;
+        }
+        // Dropped even if their writing fails: they would be written again, and fail again, with the next ones
+        $records = $this->records;
+        $this->records = new ArrayCollection();
         $entities = [];
-        $currentProcessExecution = $this->records->isEmpty() ? null : $this->processExecutionManager?->getManagedProcessExecution();
-        foreach ($this->records as $record) {
-            if ($currentProcessExecution instanceof ProcessExecution) {
-                $entity = new \CleverAge\UiProcessBundle\Entity\LogRecord($record, $currentProcessExecution);
-                $this->em?->persist($entity);
-                $entities[] = $entity;
+        try {
+            $currentProcessExecution = $this->processExecutionManager?->getManagedProcessExecution();
+            foreach ($records as $record) {
+                if ($currentProcessExecution instanceof ProcessExecution) {
+                    $entity = new \CleverAge\UiProcessBundle\Entity\LogRecord($record, $currentProcessExecution);
+                    $this->em?->persist($entity);
+                    $entities[] = $entity;
+                }
             }
+        } catch (\Throwable $exception) {
+            $this->onWriteFailure($exception, $records->count(), $entities);
+
+            return;
+        }
+        $onlyLogs = $this->hasOnlyLogsToWrite($currentProcessExecution, $entities);
+        try {
+            $this->em?->flush();
+        } catch (\Throwable $exception) {
+            if (!$onlyLogs) {
+                // The entity manager is shared with the process tasks: the failure may come from their entities
+                throw $exception;
+            }
+            // The logs must neither make the process fail nor prevent saving the status of its execution
+            $this->onWriteFailure($exception, $records->count(), $entities);
+
+            return;
         }
-        $this->em?->flush();
         // Written log records are no longer needed: detached so that the identity map does not grow during long
         // processes (detach is not cascaded to the process execution, see LogRecord::$processExecution)
         foreach ($entities as $entity) {
             $this->em?->detach($entity);
         }
-        $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.
+     * Drop the records not written at the end of the previous process execution (logged after its last flush), e.g.
+     * between two messages of a long-running worker: they would be attached to the next one.
      */
     #[\Override]
     public function reset(): void
@@ -104,9 +142,97 @@ protected function write(LogRecord $record): void
         if (!$this->enabled) {
             return;
         }
+        if ($this->reportingFailure) {
+            // The failure of the writing is not stored in database
+            return;
+        }
         $this->records->add($record);
-        if (500 === $this->records->count()) {
+        if ($this->records->count() >= 500) {
             $this->flush();
         }
     }
+
+    /**
+     * Only the log records and the current process execution are to be written by the flush: no change of the entities
+     * of the process tasks, sharing the entity manager (an error of the computation of the changes, e.g. a new entity
+     * not persisted, is not caught).
+     *
+     * @param list<\CleverAge\UiProcessBundle\Entity\LogRecord> $entities
+     */
+    private function hasOnlyLogsToWrite(?ProcessExecution $currentProcessExecution, array $entities): bool
+    {
+        if (!$this->em instanceof EntityManagerInterface) {
+            return true;
+        }
+        $ownEntities = [];
+        foreach ([...$entities, $currentProcessExecution] as $entity) {
+            if (null !== $entity) {
+                $ownEntities[spl_object_id($entity)] = true;
+            }
+        }
+        $isOwnEntity = static fn (object $entity): bool => isset($ownEntities[spl_object_id($entity)]);
+        $unitOfWork = $this->em->getUnitOfWork();
+        // Known without computing the changes
+        foreach ($unitOfWork->getScheduledEntityInsertions() as $entity) {
+            if (!$isOwnEntity($entity)) {
+                return false;
+            }
+        }
+        if ([] !== $unitOfWork->getScheduledEntityDeletions() || [] !== $unitOfWork->getScheduledCollectionDeletions()) {
+            return false;
+        }
+        $onlyOwnEntitiesManaged = true;
+        foreach ($unitOfWork->getIdentityMap() as $managedEntities) {
+            foreach ($managedEntities as $entity) {
+                if (!$isOwnEntity($entity)) {
+                    $onlyOwnEntitiesManaged = false;
+                    break 2;
+                }
+            }
+        }
+        if ($onlyOwnEntitiesManaged) {
+            // No entity of the tasks to update
+            return true;
+        }
+        // The updates of the managed entities are known once the changes computed (computed again by the flush)
+        $unitOfWork->computeChangeSets();
+        foreach ($unitOfWork->getScheduledEntityUpdates() as $entity) {
+            if (!$isOwnEntity($entity)) {
+                return false;
+            }
+        }
+
+        return [] === $unitOfWork->getScheduledCollectionUpdates() && [] === $unitOfWork->getScheduledCollectionDeletions();
+    }
+
+    /**
+     * @param list<\CleverAge\UiProcessBundle\Entity\LogRecord> $entities
+     */
+    private function onWriteFailure(\Throwable $exception, int $count, array $entities): void
+    {
+        $this->reportingFailure = true;
+        try {
+            $this->logger?->error(
+                \sprintf('%d log record(s) could not be stored in database: %s', $count, $exception->getMessage()),
+                ['exception' => $exception]
+            );
+        } finally {
+            $this->reportingFailure = false;
+        }
+        if (true === $this->em?->isOpen()) {
+            // Not inserted by the next flush, e.g. when saving the process execution
+            foreach ($entities as $entity) {
+                $this->em->detach($entity);
+            }
+
+            return;
+        }
+        try {
+            // Closed by Doctrine after the failure of the flush: reset right away for the process tasks (see
+            // ProcessExecutionRepository::getManaged())
+            $this->processExecutionManager?->getManagedProcessExecution();
+        } catch (\Throwable) {
+            // E.g. the database is still unavailable: reset again before the next writing
+        }
+    }
 }

tests/DependencyInjection/CleverAgeUiProcessExtensionTest.php

@@ -145,6 +145,9 @@ public function testLoadWithDefaultConfiguration(): void
         $databaseHandler = $container->getDefinition('cleverage_ui_process.monolog_handler.doctrine_process');
         self::assertSame('Info', $databaseHandler->getArgument('$level'));
         self::assertFalse($databaseHandler->hasMethodCall('disable'));
+        // Failures to store the logs reported on the channel of the process logs (logger set by the MonologBundle)
+        self::assertTrue($databaseHandler->hasMethodCall('setLogger'));
+        self::assertSame([['channel' => 'cleverage_process']], $databaseHandler->getTag('monolog.logger'));
 
         $purgeLogsCommand = $container->getDefinition('cleverage_ui_process.command.purge_logs');
         self::assertNull($purgeLogsCommand->getArgument('$databaseMaxDays'));