Merge pull request #5392 from mhsdesign/task/subscription-engine-save…

…point-simplication

TASK: Overhaul CatchUpHook error behaviour; At least once delivery for ERROR projections

Neos.ContentGraph.DoctrineDbalAdapter/src/DoctrineDbalContentGraphProjection.php

@@ -62,7 +62,6 @@
 use Neos\ContentRepository\Core\Feature\WorkspaceRebase\Event\WorkspaceRebaseFailed;
 use Neos\ContentRepository\Core\Feature\WorkspaceRebase\Event\WorkspaceWasRebased;
 use Neos\ContentRepository\Core\Infrastructure\DbalSchemaDiff;
-use Neos\ContentRepository\Core\Infrastructure\ProjectionTransactionTrait;
 use Neos\ContentRepository\Core\NodeType\NodeTypeName;
 use Neos\ContentRepository\Core\Projection\ContentGraph\ContentGraphProjectionInterface;
 use Neos\ContentRepository\Core\Projection\ContentGraph\ContentGraphReadModelInterface;
@@ -81,8 +80,6 @@
  */
 final class DoctrineDbalContentGraphProjection implements ContentGraphProjectionInterface
 {
-    use ProjectionTransactionTrait;
-
     use ContentStream;
     use NodeMove;
     use NodeRemoval;

Neos.ContentGraph.PostgreSQLAdapter/src/Domain/Projection/HypergraphProjection.php

@@ -40,7 +40,6 @@
 use Neos\ContentRepository\Core\Feature\SubtreeTagging\Event\SubtreeWasTagged;
 use Neos\ContentRepository\Core\Feature\SubtreeTagging\Event\SubtreeWasUntagged;
 use Neos\ContentRepository\Core\Infrastructure\DbalSchemaDiff;
-use Neos\ContentRepository\Core\Infrastructure\ProjectionTransactionTrait;
 use Neos\ContentRepository\Core\Projection\ContentGraph\ContentGraphProjectionInterface;
 use Neos\ContentRepository\Core\Projection\ContentGraph\ContentGraphReadModelInterface;
 use Neos\ContentRepository\Core\Projection\ProjectionStatus;
@@ -53,8 +52,6 @@
  */
 final class HypergraphProjection implements ContentGraphProjectionInterface
 {
-    use ProjectionTransactionTrait;
-
     use ContentStreamForking;
     use NodeCreation;
     use SubtreeTagging;

Neos.ContentRepository.BehavioralTests/Classes/ProjectionRaceConditionTester/RaceTrackerCatchUpHook.php

@@ -127,6 +127,10 @@ public function onAfterEvent(EventInterface $eventInstance, EventEnvelope $event
     {
     }
 
+    public function onAfterBatchCompleted(): void
+    {
+    }
+
     public function onAfterCatchUp(): void
     {
         // we only want to track relevant lock release calls (i.e. if we were in the event processing loop before)

Neos.ContentRepository.BehavioralTests/Classes/TestSuite/DebugEventProjection.php

@@ -12,7 +12,6 @@
 use Neos\ContentRepository\Core\EventStore\EventInterface;
 use Neos\ContentRepository\Core\Infrastructure\DbalSchemaDiff;
 use Neos\ContentRepository\Core\Infrastructure\DbalSchemaFactory;
-use Neos\ContentRepository\Core\Infrastructure\ProjectionTransactionTrait;
 use Neos\ContentRepository\Core\Projection\ProjectionInterface;
 use Neos\ContentRepository\Core\Projection\ProjectionStateInterface;
 use Neos\ContentRepository\Core\Projection\ProjectionStatus;
@@ -30,8 +29,6 @@
  */
 final class DebugEventProjection implements ProjectionInterface
 {
-    use ProjectionTransactionTrait;
-
     private DebugEventProjectionState $state;
 
     private \Closure|null $saboteur = null;
@@ -53,6 +50,9 @@ public function setUp(): void
         foreach ($this->determineRequiredSqlStatements() as $statement) {
             $this->dbal->executeStatement($statement);
         }
+        if ($this->saboteur) {
+            ($this->saboteur)();
+        }
     }
 
     public function status(): ProjectionStatus

Neos.ContentRepository.BehavioralTests/Classes/TestSuite/DebugEventProjectionState.php

@@ -21,12 +21,23 @@ public function __construct(
     }
 
     /**
-     * @return iterable<SequenceNumber>
+     * @return array<SequenceNumber>
      */
-    public function findAppliedSequenceNumbers(): iterable
+    public function findAppliedSequenceNumbers(): array
     {
         return array_map(
-            fn ($value) => SequenceNumber::fromInteger((int)$value['sequenceNumber']),
+            fn (int $value) => SequenceNumber::fromInteger($value),
+            $this->findAppliedSequenceNumberValues()
+        );
+    }
+
+    /**
+     * @return array<int>
+     */
+    public function findAppliedSequenceNumberValues(): array
+    {
+        return array_map(
+            fn ($value) => (int)$value['sequenceNumber'],
             $this->dbal->fetchAllAssociative("SELECT sequenceNumber from {$this->tableNamePrefix}")
         );
     }

Neos.ContentRepository.BehavioralTests/Tests/Functional/Subscription/AbstractSubscriptionEngineTestCase.php

@@ -44,6 +44,8 @@
  */
 abstract class AbstractSubscriptionEngineTestCase extends TestCase // we don't use Flows functional test case as it would reset the database afterwards
 {
+    protected static ContentRepositoryId $contentRepositoryId;
+
     protected ContentRepository $contentRepository;
 
     protected SubscriptionEngine $subscriptionEngine;
@@ -56,32 +58,37 @@ abstract class AbstractSubscriptionEngineTestCase extends TestCase // we don't u
 
     protected CatchUpHookInterface&MockObject $catchupHookForFakeProjection;
 
+    public static function setUpBeforeClass(): void
+    {
+        static::$contentRepositoryId = ContentRepositoryId::fromString('t_subscription');
+    }
+
     public function setUp(): void
     {
         if ($this->getObject(Connection::class)->getDatabasePlatform() instanceof PostgreSQLPlatform) {
             $this->markTestSkipped('TODO: The content graph is not available in postgres currently: https://github.com/neos/neos-development-collection/issues/3855');
         }
-        $contentRepositoryId = ContentRepositoryId::fromString('t_subscription');
 
         $this->resetDatabase(
             $this->getObject(Connection::class),
-            $contentRepositoryId,
+            self::$contentRepositoryId,
             keepSchema: true
         );
 
         $this->fakeProjection = $this->getMockBuilder(ProjectionInterface::class)->disableAutoReturnValueGeneration()->getMock();
         $this->fakeProjection->method('getState')->willReturn(new class implements ProjectionStateInterface {});
-        $this->fakeProjection->expects(self::any())->method('transactional')->willReturnCallback(fn ($fn) => $fn())->willReturnCallback(fn ($fn) => $fn());
 
         FakeProjectionFactory::setProjection(
             'default',
             $this->fakeProjection
         );
 
-        $this->secondFakeProjection = new DebugEventProjection(
-            sprintf('cr_%s_debug_projection', $contentRepositoryId->value),
-            $this->getObject(Connection::class)
-        );
+        if (!isset($this->secondFakeProjection)) {
+            $this->secondFakeProjection = new DebugEventProjection(
+                sprintf('cr_%s_debug_projection', self::$contentRepositoryId->value),
+                $this->getObject(Connection::class)
+            );
+        }
 
         FakeProjectionFactory::setProjection(
             'second',
@@ -98,9 +105,9 @@ public function setUp(): void
         FakeNodeTypeManagerFactory::setConfiguration([]);
         FakeContentDimensionSourceFactory::setWithoutDimensions();
 
-        $this->getObject(ContentRepositoryRegistry::class)->resetFactoryInstance($contentRepositoryId);
+        $this->getObject(ContentRepositoryRegistry::class)->resetFactoryInstance(self::$contentRepositoryId);
 
-        $this->setupContentRepositoryDependencies($contentRepositoryId);
+        $this->setupContentRepositoryDependencies(self::$contentRepositoryId);
     }
 
     final protected function setupContentRepositoryDependencies(ContentRepositoryId $contentRepositoryId)
@@ -180,7 +187,7 @@ final protected function commitExampleContentStreamEvent(): void
         );
     }
 
-    final protected function expectOkayStatus($subscriptionId, SubscriptionStatus $status, SequenceNumber $sequenceNumber): void
+    final protected function expectOkayStatus(string $subscriptionId, SubscriptionStatus $status, SequenceNumber $sequenceNumber): void
     {
         $actual = $this->subscriptionStatus($subscriptionId);
         self::assertEquals(