keep running projections

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/PdoEventStoreReadModelProjectorTestCase.php

@@ -14,6 +14,7 @@
 namespace ProophTest\EventStore\Pdo\Projection;
 
 use ArrayIterator;
+use Assert\Assertion;
 use DateInterval;
 use DateTimeImmutable;
 use DateTimeZone;
@@ -26,6 +27,7 @@
 use Prooph\EventStore\Pdo\Projection\PdoEventStoreProjector;
 use Prooph\EventStore\Pdo\Projection\PdoEventStoreReadModelProjector;
 use Prooph\EventStore\Projection\ProjectionManager;
+use Prooph\EventStore\Projection\ProjectionStatus;
 use Prooph\EventStore\Projection\Projector;
 use Prooph\EventStore\Projection\ReadModel;
 use Prooph\EventStore\Stream;
@@ -657,6 +659,68 @@ public function it_does_not_perform_retry_when_event_is_older_than_detection_win
         $this->assertFalse($gapDetection->isRetrying());
     }
 
+    /**
+     * @test
+     */
+    public function projection_should_run_until_end_of_stream(): void
+    {
+        $this->prepareEventStream('user-345');
+
+        $projectionManager = $this->projectionManager;
+        $projection = $projectionManager->createReadModelProjection('test_projection', new ReadModelMock(), [
+            Projector::OPTION_PERSIST_BLOCK_SIZE => 1,
+            PdoEventStoreReadModelProjector::OPTION_LOAD_COUNT => 1,
+        ]);
+
+        $projection
+            ->fromStream('user-345')
+            ->whenAny(function () use ($projectionManager) {
+                Assertion::eq(ProjectionStatus::RUNNING(), $projectionManager->fetchProjectionStatus('test_projection'));
+            })
+            ->run(false);
+
+        $this->assertEquals(50, $projectionManager->fetchProjectionStreamPositions('test_projection')['user-345']);
+        $this->assertEquals(ProjectionStatus::IDLE(), $projectionManager->fetchProjectionStatus('test_projection'));
+    }
+
+    /**
+     * @test
+     */
+    public function when_failed_projection_should_release_lock_but_indicate_running_status(): void
+    {
+        $this->prepareEventStream('user-345');
+
+        $projectionManager = $this->projectionManager;
+        $projection = $projectionManager->createReadModelProjection('test_projection', new ReadModelMock(), [
+            Projector::OPTION_PERSIST_BLOCK_SIZE => 1,
+            PdoEventStoreReadModelProjector::OPTION_LOAD_COUNT => 1,
+        ]);
+
+        $projection
+            ->fromStream('user-345')
+            ->init(function () {
+                return ['iteration' => 0];
+            })
+            ->whenAny(function (array $state, Message $event): array {
+                ++$state['iteration'];
+
+                if ($state['iteration'] > 5) {
+                    throw new \RuntimeException('something happened');
+                }
+
+                return $state;
+            });
+
+        try {
+            $projection->run(false);
+        } catch (\Throwable) {
+        }
+
+        $this->assertEquals(5, $projectionManager->fetchProjectionStreamPositions('test_projection')['user-345']);
+        $this->assertEquals(ProjectionStatus::RUNNING(), $projectionManager->fetchProjectionStatus('test_projection'));
+        $this->assertNull($this->connection->query("select locked_until from projections where name = 'test_projection'")->fetch(PDO::FETCH_COLUMN));
+    }
+
     protected function prepareEventStreamWithOneEvent(string $name, ?DateTimeImmutable $createdAt = null): void
     {
         $events = [];