Bugs fix

CHANGELOG.md

@@ -1,3 +1,14 @@
+# Version 6.0.3
+* Fix job error count display using `Job::getExceptionCount()` when available in `JobShowCommand`
+* Fix exceptions retrieval when only a stream is available in `JobShowCommand`
+* Fix `JobMessageHandler` skipping processing when the job can no longer be found
+* Fix temp log file not being removed when an exception occurs during processing
+* Fix exceptions loop stopping on a falsy (`"0"`) log line in `JobProcessor`
+* Fix array cast in `InitFromDbTrait`
+
+# Version 6.0.2
+* Fix null stream causing an error when no exceptions occured in `JobProcessor`
+
 # Version 6.0.1
 * Fix esource can be null
 

src/Command/JobShowCommand.php

@@ -72,7 +72,7 @@ public function __invoke(
             ['Started at', $job->getStartTime() ? $job->getStartTime()->format('Y-m-d H:i:s') : '-'],
             ['Ended at', $job->getEndTime() ? $job->getEndTime()->format('Y-m-d H:i:s') : '-'],
             ['Object number', $job->getCount()],
-            ['Errors', \count((array) $job->getExceptions())],
+            ['Errors', $job->getExceptionCount() ?? \count((array) $job->getExceptions())],
             ['Status', $this->translateStatus($job->getStatus())],
         ];
         if ($details) {
@@ -84,9 +84,7 @@ public function __invoke(
         $io->table(['Field', 'Value'], $display);
         if ($details) {
             $io->section('Exceptions');
-            $exceptions = array_map(static fn (string $exception) => substr($exception, 0, 900).'…', $job->getExceptions());
-
-            $io->write($exceptions);
+            $io->write($this->resolveExceptionLines($job));
         }
 
         return 0;
@@ -96,4 +94,22 @@ private function translateStatus(int $status): string
     {
         return self::STATUS_MAPPING[$status] ?? 'Unknown status';
     }
+
+    /**
+     * @return string[]
+     */
+    private function resolveExceptionLines(Job $job): array
+    {
+        $exceptions = $job->getExceptions();
+
+        if ($exceptions === null && \is_resource($stream = $job->getStreamExceptions())) {
+            rewind($stream);
+            $exceptions = [];
+            while (false !== ($line = fgets($stream))) {
+                $exceptions[] = $line;
+            }
+        }
+
+        return array_map(static fn (string $exception) => substr($exception, 0, 900).'…', $exceptions ?? []);
+    }
 }

src/DataflowType/Result.php

@@ -13,6 +13,9 @@ class Result
 
     private int $successCount;
 
+    /** @var array<int, mixed> */
+    private array $exceptions = [];
+
     public function __construct(private readonly string $name, private readonly \DateTimeInterface $startTime, private readonly \DateTimeInterface $endTime, private readonly int $totalProcessedCount, private readonly int $errorCount)
     {
         $this->elapsed = $startTime->diff($endTime);

src/MessengerMode/JobMessageHandler.php

@@ -17,6 +17,11 @@ public function __construct(private readonly JobRepository $repository, private
 
     public function __invoke(JobMessage $message): void
     {
-        $this->processor->process($this->repository->find($message->getJobId()));
+        $job = $this->repository->find($message->getJobId());
+        if ($job === null) {
+            return;
+        }
+
+        $this->processor->process($job);
     }
 }

src/Processor/JobProcessor.php

@@ -45,7 +45,8 @@ public function process(Job $job): void
             $dataflowType->setRepository($this->repository);
         }
 
-        $handler = new StreamHandler(tempnam(sys_get_temp_dir(), 'dataflow_'), fileOpenMode: 'w+');
+        $tempFile = tempnam(sys_get_temp_dir(), 'dataflow_');
+        $handler = new StreamHandler($tempFile, fileOpenMode: 'w+');
         $handler->setFormatter(new LineFormatter(self::FORMAT));
 
         $loggers = [new Logger('dataflow_internal', [$bufferHandler = $handler])];
@@ -56,9 +57,15 @@ public function process(Job $job): void
 
         $dataflowType->setLogger($logger);
 
-        $result = $dataflowType->process($job->getOptions(), $job->getId());
+        try {
+            $result = $dataflowType->process($job->getOptions(), $job->getId());
 
-        $this->afterProcessing($job, $result, $bufferHandler);
+            $this->afterProcessing($job, $result, $bufferHandler);
+        } finally {
+            if (is_file($tempFile)) {
+                @unlink($tempFile);
+            }
+        }
     }
 
     private function beforeProcessing(Job $job): void
@@ -89,7 +96,7 @@ private function afterProcessing(Job $job, Result $result, StreamHandler $stream
             $exceptions = [];
             if ($stream !== null) {
                 rewind($stream);
-                while ($line = fgets($stream)) {
+                while (false !== ($line = fgets($stream))) {
                     $exceptions[] = $line;
                 }
             }

src/Repository/InitFromDbTrait.php

@@ -42,6 +42,6 @@ private function strToArray($value): array
 
         $array = json_decode($value, true, 512, \JSON_THROW_ON_ERROR);
 
-        return ($array === false) ? [] : $array;
+        return \is_array($array) ? $array : [];
     }
 }