keep projections running as long events are loading

projections should keep running when:
- runs with `keepRunning: true` due it awaits for new events
- it did not reached end of the stream, e.g. in case of resetting

projection status should indicate actual status of projection therefore, `idle` should be set if projection wasn't supposed to keep running or end of stream was reached

Author: unixslayer

src/Projection/PdoEventStoreReadModelProjector.php

@@ -132,11 +132,6 @@ final class PdoEventStoreReadModelProjector implements MetadataAwareReadModelPro
      */
     private $eventCounter = 0;
 
-    /**
-     * @var int
-     */
-    private $loadedEvents = 0;
-
     /**
      * @var int
      */
@@ -529,7 +524,7 @@ public function run(bool $keepRunning = true): void
                 }
 
                 $streamEvents = new MergedStreamIterator(\array_keys($eventStreams), ...\array_values($eventStreams));
-                $this->loadedEvents = $streamEvents->count();
+                $loadedEvents = $streamEvents->count();
 
                 if ($singleHandler) {
                     $gapDetected = ! $this->handleStreamWithSingleHandler($streamEvents);
@@ -586,9 +581,9 @@ public function run(bool $keepRunning = true): void
                 }
 
                 $this->prepareStreamPositions();
-            } while ($keepRunning && ! $this->isStopped);
+            } while (($keepRunning || $loadedEvents > 0) && ! $this->isStopped);
         } finally {
-            $this->releaseLock($keepRunning);
+            $this->releaseLock($keepRunning, $loadedEvents);
         }
     }
 
@@ -915,7 +910,7 @@ private function updateLock(): void
         $this->lastLockUpdate = $now;
     }
 
-    private function releaseLock(bool $keepRunning): void
+    private function releaseLock(bool $keepRunning, int $loadedEvents): void
     {
         $projectionsTable = $this->quoteTableName($this->projectionsTable);
         $sql = <<<EOT
@@ -924,7 +919,7 @@ private function releaseLock(bool $keepRunning): void
 
         $statement = $this->connection->prepare($sql);
 
-        $status = $keepRunning && $this->loadedEvents > 0 ? ProjectionStatus::RUNNING() : ProjectionStatus::IDLE();
+        $status = ($keepRunning || $loadedEvents > 0) ? ProjectionStatus::RUNNING() : ProjectionStatus::IDLE();
 
         try {
             $statement->execute([$status->getValue(), $this->name]);

tests/Projection/AbstractPdoEventStoreReadModelProjectorTestCase.php

@@ -14,6 +14,7 @@
 namespace ProophTest\EventStore\Pdo\Projection;
 
 use ArrayIterator;
+use Assert\Assertion;
 use DateInterval;
 use DateTimeImmutable;
 use DateTimeZone;
@@ -27,6 +28,7 @@
 use Prooph\EventStore\Pdo\Projection\GapDetection;
 use Prooph\EventStore\Pdo\Projection\PdoEventStoreProjector;
 use Prooph\EventStore\Pdo\Projection\PdoEventStoreReadModelProjector;
+use Prooph\EventStore\Projection\ProjectionStatus;
 use Prooph\EventStore\Projection\Projector;
 use Prooph\EventStore\Projection\ReadModel;
 use Prooph\EventStore\Stream;
@@ -456,27 +458,21 @@ public function it_detects_gap_and_performs_retry(): void
         $projection
             ->fromStream('user')
             ->init(function () {
-                return [];
+                return ['iteration' => 0];
             })
-            ->when([
-                UserCreated::class => function (array $state, Message $event): array {
-                    return $state;
-                },
-                UsernameChanged::class => function (array $state, Message $event): array {
-                    return $state;
-                },
-            ])
-            ->run(false);
-
-        $this->assertEquals(1, $projectionManager->fetchProjectionStreamPositions('test_projection')['user']);
-
-        $this->assertTrue($gapDetection->isRetrying());
+            ->whenAny(
+                function (array $state, Message $event) use ($projectionManager, $gapDetection, $parallelConnection): array {
+                    if ($state['iteration'] === 1) {
+                        Assertion::true($gapDetection->isRetrying());
+                        $parallelConnection->commit();
+                    }
 
-        // Fill the gap
-        $parallelConnection->commit();
+                    ++$state['iteration'];
 
-        // Run again with gap detection in retry mode
-        $projection->run(false);
+                    return $state;
+                }
+            )
+            ->run(false);
 
         $this->assertEquals(3, $projectionManager->fetchProjectionStreamPositions('test_projection')['user']);
 
@@ -521,36 +517,25 @@ public function it_continues_when_retry_limit_is_reached_and_gap_not_filled(): v
         $projection
             ->fromStream('user')
             ->init(function () {
-                return [];
+                return ['iteration' => 0];
             })
-            ->when([
-                UserCreated::class => function (array $state, Message $event): array {
-                    return $state;
-                },
-                UsernameChanged::class => function (array $state, Message $event): array {
-                    return $state;
-                },
-            ])
-            ->run(false);
-
-        $this->assertEquals(1, $projectionManager->fetchProjectionStreamPositions('test_projection')['user']);
-
-        $this->assertTrue($gapDetection->isRetrying());
-
-        // Force a real gap
-        $parallelConnection->rollBack();
-
-        // Run again with gap detection in retry mode
-        $projection->run(false);
+            ->whenAny(
+                function (array $state, Message $event) use ($projectionManager, $gapDetection, $parallelConnection): array {
+                    if ($state['iteration'] === 1) {
+                        Assertion::true($gapDetection->isRetrying());
+                        $parallelConnection->rollBack();
+                    }
 
-        // Projection should not move forward, but instead retry a second time
-        $this->assertEquals(1, $projectionManager->fetchProjectionStreamPositions('test_projection')['user']);
+                    ++$state['iteration'];
 
-        // Third run with gap detection still in retry mode, but limit reached
-        $projection->run(false);
+                    return $state;
+                }
+            )
+            ->run(false);
 
         //Projection should have moved forward
         $this->assertEquals(3, $projectionManager->fetchProjectionStreamPositions('test_projection')['user']);
+        $this->assertEquals(ProjectionStatus::IDLE(), $projectionManager->fetchProjectionStatus('test_projection'));
 
         $this->assertFalse($gapDetection->isRetrying());
     }