Merge pull request #5297 from mhsdesign/bugfix/harden-content-stream-…

…pruner

BUGFIX: Harden content stream pruner and remove `removed` flag from content streams

Neos.ContentGraph.DoctrineDbalAdapter/src/ContentGraphReadModelAdapter.php

@@ -26,7 +26,6 @@
 use Neos\ContentRepository\Core\SharedModel\Workspace\ContentStream;
 use Neos\ContentRepository\Core\SharedModel\Workspace\ContentStreamId;
 use Neos\ContentRepository\Core\SharedModel\Workspace\ContentStreams;
-use Neos\ContentRepository\Core\SharedModel\Workspace\ContentStreamStatus;
 use Neos\ContentRepository\Core\SharedModel\Workspace\Workspace;
 use Neos\ContentRepository\Core\SharedModel\Workspace\WorkspaceName;
 use Neos\ContentRepository\Core\SharedModel\Workspace\Workspaces;
@@ -104,7 +103,7 @@ public function findContentStreamById(ContentStreamId $contentStreamId): ?Conten
     {
         $contentStreamByIdStatement = <<<SQL
             SELECT
-                id, sourceContentStreamId, status, version, removed
+                id, sourceContentStreamId, version, closed
             FROM
                 {$this->tableNames->contentStream()}
             WHERE
@@ -128,7 +127,7 @@ public function findContentStreams(): ContentStreams
     {
         $contentStreamsStatement = <<<SQL
             SELECT
-                id, sourceContentStreamId, status, version, removed
+                id, sourceContentStreamId, version, closed
             FROM
                 {$this->tableNames->contentStream()}
         SQL;
@@ -160,7 +159,7 @@ private function getBasicWorkspaceQuery(): QueryBuilder
         $queryBuilder = $this->dbal->createQueryBuilder();
 
         return $queryBuilder
-            ->select('ws.name, ws.baseWorkspaceName, ws.currentContentStreamId, cs.sourceContentStreamVersion != scs.version as baseWorkspaceChanged')
+            ->select('ws.name, ws.baseWorkspaceName, ws.currentContentStreamId, cs.sourceContentStreamVersion = scs.version as upToDateWithBase')
             ->from($this->tableNames->workspace(), 'ws')
             ->join('ws', $this->tableNames->contentStream(), 'cs', 'cs.id = ws.currentcontentstreamid')
             ->leftJoin('cs', $this->tableNames->contentStream(), 'scs', 'scs.id = cs.sourceContentStreamId');
@@ -172,15 +171,19 @@ private function getBasicWorkspaceQuery(): QueryBuilder
     private static function workspaceFromDatabaseRow(array $row): Workspace
     {
         $baseWorkspaceName = $row['baseWorkspaceName'] !== null ? WorkspaceName::fromString($row['baseWorkspaceName']) : null;
-        $status = match ($row['baseWorkspaceChanged']) {
+
+        if ($baseWorkspaceName === null) {
             // no base workspace, a root is always up-to-date
-            null => WorkspaceStatus::UP_TO_DATE,
-            // base workspace didnt change (sql 0 is _false_)
-            0 => WorkspaceStatus::UP_TO_DATE,
-            default => WorkspaceStatus::OUTDATED,
-        };
+            $status = WorkspaceStatus::UP_TO_DATE;
+        } elseif ($row['upToDateWithBase'] === 1) {
+            // base workspace didnt change
+            $status = WorkspaceStatus::UP_TO_DATE;
+        } else {
+            // base content stream was removed or contains newer changes
+            $status = WorkspaceStatus::OUTDATED;
+        }
 
-        return new Workspace(
+        return Workspace::create(
             WorkspaceName::fromString($row['name']),
             $baseWorkspaceName,
             ContentStreamId::fromString($row['currentContentStreamId']),
@@ -193,12 +196,11 @@ private static function workspaceFromDatabaseRow(array $row): Workspace
      */
     private static function contentStreamFromDatabaseRow(array $row): ContentStream
     {
-        return new ContentStream(
+        return ContentStream::create(
             ContentStreamId::fromString($row['id']),
             isset($row['sourceContentStreamId']) ? ContentStreamId::fromString($row['sourceContentStreamId']) : null,
-            ContentStreamStatus::from($row['status']),
             Version::fromInteger((int)$row['version']),
-            (bool)$row['removed']
+            (bool)$row['closed'],
         );
     }
 }

Neos.ContentGraph.DoctrineDbalAdapter/src/DoctrineDbalContentGraphProjection.php

@@ -17,12 +17,11 @@
 use Neos\ContentGraph\DoctrineDbalAdapter\Domain\Projection\NodeRelationAnchorPoint;
 use Neos\ContentGraph\DoctrineDbalAdapter\Domain\Repository\DimensionSpacePointsRepository;
 use Neos\ContentGraph\DoctrineDbalAdapter\Domain\Repository\ProjectionContentGraph;
-use Neos\ContentRepository\Core\EventStore\InitiatingEventMetadata;
-use Neos\ContentRepository\Core\Projection\ContentGraph\ContentGraphReadModelInterface;
 use Neos\ContentRepository\Core\DimensionSpace\DimensionSpacePoint;
 use Neos\ContentRepository\Core\DimensionSpace\DimensionSpacePointSet;
 use Neos\ContentRepository\Core\DimensionSpace\OriginDimensionSpacePoint;
 use Neos\ContentRepository\Core\EventStore\EventInterface;
+use Neos\ContentRepository\Core\EventStore\InitiatingEventMetadata;
 use Neos\ContentRepository\Core\Feature\Common\EmbedsContentStreamId;
 use Neos\ContentRepository\Core\Feature\Common\InterdimensionalSiblings;
 use Neos\ContentRepository\Core\Feature\ContentStreamClosing\Event\ContentStreamWasClosed;
@@ -64,15 +63,15 @@
 use Neos\ContentRepository\Core\Infrastructure\DbalSchemaDiff;
 use Neos\ContentRepository\Core\NodeType\NodeTypeName;
 use Neos\ContentRepository\Core\Projection\CheckpointStorageStatusType;
+use Neos\ContentRepository\Core\Projection\ContentGraph\ContentGraphProjectionInterface;
+use Neos\ContentRepository\Core\Projection\ContentGraph\ContentGraphReadModelInterface;
 use Neos\ContentRepository\Core\Projection\ContentGraph\NodeTags;
 use Neos\ContentRepository\Core\Projection\ContentGraph\Timestamps;
 use Neos\ContentRepository\Core\Projection\ProjectionStatus;
-use Neos\ContentRepository\Core\Projection\ContentGraph\ContentGraphProjectionInterface;
 use Neos\ContentRepository\Core\SharedModel\Node\NodeAggregateClassification;
 use Neos\ContentRepository\Core\SharedModel\Node\NodeAggregateId;
 use Neos\ContentRepository\Core\SharedModel\Node\NodeName;
 use Neos\ContentRepository\Core\SharedModel\Workspace\ContentStreamId;
-use Neos\ContentRepository\Core\SharedModel\Workspace\ContentStreamStatus;
 use Neos\EventStore\Model\Event\SequenceNumber;
 use Neos\EventStore\Model\EventEnvelope;
 
@@ -259,12 +258,12 @@ public function inSimulation(\Closure $fn): mixed
 
     private function whenContentStreamWasClosed(ContentStreamWasClosed $event): void
     {
-        $this->updateContentStreamStatus($event->contentStreamId, ContentStreamStatus::CLOSED);
+        $this->closeContentStream($event->contentStreamId);
     }
 
     private function whenContentStreamWasCreated(ContentStreamWasCreated $event): void
     {
-        $this->createContentStream($event->contentStreamId, ContentStreamStatus::CREATED);
+        $this->createContentStream($event->contentStreamId);
     }
 
     private function whenContentStreamWasForked(ContentStreamWasForked $event): void
@@ -303,7 +302,7 @@ private function whenContentStreamWasForked(ContentStreamWasForked $event): void
         // NOTE: as reference edges are attached to Relation Anchor Points (and they are lazily copy-on-written),
         // we do not need to copy reference edges here (but we need to do it during copy on write).
 
-        $this->createContentStream($event->newContentStreamId, ContentStreamStatus::FORKED, $event->sourceContentStreamId, $event->versionOfSourceContentStream);
+        $this->createContentStream($event->newContentStreamId, $event->sourceContentStreamId, $event->versionOfSourceContentStream);
     }
 
     private function whenContentStreamWasRemoved(ContentStreamWasRemoved $event): void
@@ -353,7 +352,7 @@ private function whenContentStreamWasRemoved(ContentStreamWasRemoved $event): vo
 
     private function whenContentStreamWasReopened(ContentStreamWasReopened $event): void
     {
-        $this->updateContentStreamStatus($event->contentStreamId, $event->previousState);
+        $this->reopenContentStream($event->contentStreamId);
     }
 
     private function whenDimensionShineThroughWasAdded(DimensionShineThroughWasAdded $event): void
@@ -708,9 +707,6 @@ private function whenRootNodeAggregateWithNodeWasCreated(RootNodeAggregateWithNo
     private function whenRootWorkspaceWasCreated(RootWorkspaceWasCreated $event): void
     {
         $this->createWorkspace($event->workspaceName, null, $event->newContentStreamId);
-
-        // the content stream is in use now
-        $this->updateContentStreamStatus($event->newContentStreamId, ContentStreamStatus::IN_USE_BY_WORKSPACE);
     }
 
     private function whenSubtreeWasTagged(SubtreeWasTagged $event): void
@@ -730,69 +726,41 @@ private function whenWorkspaceBaseWorkspaceWasChanged(WorkspaceBaseWorkspaceWasC
 
     private function whenWorkspaceRebaseFailed(WorkspaceRebaseFailed $event): void
     {
-        $this->updateContentStreamStatus($event->candidateContentStreamId, ContentStreamStatus::REBASE_ERROR);
+        // legacy handling:
+        // before https://github.com/neos/neos-development-collection/pull/4965 this event was emitted and set the content stream status to `REBASE_ERROR`
+        // instead of setting the error state on replay for old events we make it almost behave like if the rebase had failed today: reopen the workspaces content stream id
+        // the candidateContentStreamId will be removed by the ContentStreamPruner
+        $this->reopenContentStream($event->sourceContentStreamId);
     }
 
     private function whenWorkspaceWasCreated(WorkspaceWasCreated $event): void
     {
         $this->createWorkspace($event->workspaceName, $event->baseWorkspaceName, $event->newContentStreamId);
-
-        // the content stream is in use now
-        $this->updateContentStreamStatus($event->newContentStreamId, ContentStreamStatus::IN_USE_BY_WORKSPACE);
     }
 
     private function whenWorkspaceWasDiscarded(WorkspaceWasDiscarded $event): void
     {
         $this->updateWorkspaceContentStreamId($event->workspaceName, $event->newContentStreamId);
-
-        // the new content stream is in use now
-        $this->updateContentStreamStatus($event->newContentStreamId, ContentStreamStatus::IN_USE_BY_WORKSPACE);
-        // the previous content stream is no longer in use
-        $this->updateContentStreamStatus($event->previousContentStreamId, ContentStreamStatus::NO_LONGER_IN_USE);
     }
 
     private function whenWorkspaceWasPartiallyDiscarded(WorkspaceWasPartiallyDiscarded $event): void
     {
         $this->updateWorkspaceContentStreamId($event->workspaceName, $event->newContentStreamId);
-
-        // the new content stream is in use now
-        $this->updateContentStreamStatus($event->newContentStreamId, ContentStreamStatus::IN_USE_BY_WORKSPACE);
-
-        // the previous content stream is no longer in use
-        $this->updateContentStreamStatus($event->previousContentStreamId, ContentStreamStatus::NO_LONGER_IN_USE);
     }
 
     private function whenWorkspaceWasPartiallyPublished(WorkspaceWasPartiallyPublished $event): void
     {
         $this->updateWorkspaceContentStreamId($event->sourceWorkspaceName, $event->newSourceContentStreamId);
-
-        // the new content stream is in use now
-        $this->updateContentStreamStatus($event->newSourceContentStreamId, ContentStreamStatus::IN_USE_BY_WORKSPACE);
-
-        // the previous content stream is no longer in use
-        $this->updateContentStreamStatus($event->previousSourceContentStreamId, ContentStreamStatus::NO_LONGER_IN_USE);
     }
 
     private function whenWorkspaceWasPublished(WorkspaceWasPublished $event): void
     {
         $this->updateWorkspaceContentStreamId($event->sourceWorkspaceName, $event->newSourceContentStreamId);
-
-        // the new content stream is in use now
-        $this->updateContentStreamStatus($event->newSourceContentStreamId, ContentStreamStatus::IN_USE_BY_WORKSPACE);
-
-        // the previous content stream is no longer in use
-        $this->updateContentStreamStatus($event->previousSourceContentStreamId, ContentStreamStatus::NO_LONGER_IN_USE);
     }
 
     private function whenWorkspaceWasRebased(WorkspaceWasRebased $event): void
     {
         $this->updateWorkspaceContentStreamId($event->workspaceName, $event->newContentStreamId);
-
-        // the new content stream is in use now
-        $this->updateContentStreamStatus($event->newContentStreamId, ContentStreamStatus::IN_USE_BY_WORKSPACE);
-
-        // the previous content stream is no longer in use
-        $this->updateContentStreamStatus($event->previousContentStreamId, ContentStreamStatus::NO_LONGER_IN_USE);
     }
 
     private function whenWorkspaceWasRemoved(WorkspaceWasRemoved $event): void

Neos.ContentGraph.DoctrineDbalAdapter/src/DoctrineDbalContentGraphSchemaBuilder.php

@@ -129,9 +129,7 @@ private function createContentStreamTable(): Table
             (new Column('version', Type::getType(Types::INTEGER)))->setNotnull(true),
             DbalSchemaFactory::columnForContentStreamId('sourceContentStreamId')->setNotnull(false),
             (new Column('sourceContentStreamVersion', Type::getType(Types::INTEGER)))->setNotnull(false),
-            // Should become a DB ENUM (unclear how to configure with DBAL) or int (latter needs adaption to code)
-            (new Column('status', Type::getType(Types::BINARY)))->setLength(20)->setNotnull(true),
-            (new Column('removed', Type::getType(Types::BOOLEAN)))->setDefault(false)->setNotnull(false),
+            (new Column('closed', Type::getType(Types::BOOLEAN)))->setDefault(false)->setNotnull(true),
         ]);
 
         return $contentStreamTable->setPrimaryKey(['id']);

Neos.ContentGraph.DoctrineDbalAdapter/src/Domain/Projection/Feature/ContentStream.php

@@ -5,7 +5,6 @@
 namespace Neos\ContentGraph\DoctrineDbalAdapter\Domain\Projection\Feature;
 
 use Neos\ContentRepository\Core\SharedModel\Workspace\ContentStreamId;
-use Neos\ContentRepository\Core\SharedModel\Workspace\ContentStreamStatus;
 use Neos\EventStore\Model\Event\Version;
 
 /**
@@ -15,35 +14,41 @@
  */
 trait ContentStream
 {
-    private function createContentStream(ContentStreamId $contentStreamId, ContentStreamStatus $status, ?ContentStreamId $sourceContentStreamId = null, ?Version $sourceVersion = null): void
+    private function createContentStream(ContentStreamId $contentStreamId, ?ContentStreamId $sourceContentStreamId = null, ?Version $sourceVersion = null): void
     {
         $this->dbal->insert($this->tableNames->contentStream(), [
             'id' => $contentStreamId->value,
             'version' => 0,
             'sourceContentStreamId' => $sourceContentStreamId?->value,
-            'sourceContentStreamVersion' => $sourceVersion?->value,
-            'status' => $status->value,
+            'sourceContentStreamVersion' => $sourceVersion?->value
         ]);
     }
 
-    private function updateContentStreamStatus(ContentStreamId $contentStreamId, ContentStreamStatus $status): void
+    private function closeContentStream(ContentStreamId $contentStreamId): void
     {
         $this->dbal->update($this->tableNames->contentStream(), [
-            'status' => $status->value,
+            'closed' => 1,
         ], [
             'id' => $contentStreamId->value
         ]);
     }
 
-    private function removeContentStream(ContentStreamId $contentStreamId): void
+    private function reopenContentStream(ContentStreamId $contentStreamId): void
     {
         $this->dbal->update($this->tableNames->contentStream(), [
-            'removed' => true,
+            'closed' => 0,
         ], [
             'id' => $contentStreamId->value
         ]);
     }
 
+    private function removeContentStream(ContentStreamId $contentStreamId): void
+    {
+        $this->dbal->delete($this->tableNames->contentStream(), [
+            'id' => $contentStreamId->value
+        ]);
+    }
+
     private function updateContentStreamVersion(ContentStreamId $contentStreamId, Version $version): void
     {
         $this->dbal->update($this->tableNames->contentStream(), [

Neos.ContentRepository.BehavioralTests/Tests/Behavior/Features/01-RootNodeCreation/01-CreateRootNodeAggregateWithNode_ConstraintChecks.feature

@@ -39,7 +39,7 @@ Feature: Create a root node aggregate
     Then the last command should have thrown an exception of type "WorkspaceDoesNotExist"
 
   Scenario: Try to create a root node aggregate in a closed content stream:
-    When the command CloseContentStream is executed with payload:
+    When the event ContentStreamWasClosed was published with payload:
       | Key             | Value           |
       | contentStreamId | "cs-identifier" |
     And the command CreateRootNodeAggregateWithNode is executed with payload and exceptions are caught:

Neos.ContentRepository.BehavioralTests/Tests/Behavior/Features/02-NodeCreation/01-CreateNodeAggregateWithNode_ConstraintChecks.feature

@@ -53,7 +53,7 @@ Feature: Create node aggregate with node
     Then the last command should have thrown an exception of type "WorkspaceDoesNotExist"
 
   Scenario: Try to create a node aggregate in a workspace whose content stream is closed:
-    When the command CloseContentStream is executed with payload:
+    When the event ContentStreamWasClosed was published with payload:
       | Key             | Value           |
       | contentStreamId | "cs-identifier" |
     And the command CreateNodeAggregateWithNode is executed with payload and exceptions are caught:

Neos.ContentRepository.BehavioralTests/Tests/Behavior/Features/03-NodeVariation/01-CreateNodeVariant_ConstraintChecks.feature

@@ -47,7 +47,7 @@ Feature: Create node variant
     Then the last command should have thrown an exception of type "WorkspaceDoesNotExist"
 
   Scenario: Try to create a variant in a workspace that does not exist
-    When the command CloseContentStream is executed with payload:
+    When the event ContentStreamWasClosed was published with payload:
       | Key             | Value           |
       | contentStreamId | "cs-identifier" |
     And the command CreateNodeVariant is executed with payload and exceptions are caught:

Neos.ContentRepository.BehavioralTests/Tests/Behavior/Features/04-NodeModification/01-SetNodeProperties_ConstraintChecks.feature

@@ -44,7 +44,7 @@ Feature: Set node properties: Constraint checks
     Then the last command should have thrown an exception of type "WorkspaceDoesNotExist"
 
   Scenario: Try to set properties in a workspace whose content stream is closed
-    When the command CloseContentStream is executed with payload:
+    When the event ContentStreamWasClosed was published with payload:
       | Key             | Value           |
       | contentStreamId | "cs-identifier" |
     When the command SetNodeProperties is executed with payload and exceptions are caught:

Neos.ContentRepository.BehavioralTests/Tests/Behavior/Features/05-NodeReferencing/01-SetNodeReferences_ConstraintChecks.feature

@@ -55,7 +55,7 @@ Feature: Constraint checks on SetNodeReferences
       | berta-destinode   | Neos.ContentRepository.Testing:ReferencedNode     | lady-eleonode-rootford |
 
   Scenario: Try to reference nodes in a workspace whose content stream is closed
-    When the command CloseContentStream is executed with payload:
+    When the event ContentStreamWasClosed was published with payload:
       | Key             | Value           |
       | contentStreamId | "cs-identifier" |
     When the command SetNodeReferences is executed with payload and exceptions are caught:

Neos.ContentRepository.BehavioralTests/Tests/Behavior/Features/06-NodeDisabling/01-DisableNodeAggregate_ConstraintChecks.feature

@@ -38,7 +38,7 @@ Feature: Constraint checks on node aggregate disabling
     Then the last command should have thrown an exception of type "WorkspaceDoesNotExist"
 
   Scenario: Try to disable a node aggregate in a workspace whose content stream is closed
-    When the command CloseContentStream is executed with payload:
+    When the event ContentStreamWasClosed was published with payload:
       | Key             | Value           |
       | contentStreamId | "cs-identifier" |
     When the command DisableNodeAggregate is executed with payload and exceptions are caught:

Neos.ContentRepository.BehavioralTests/Tests/Behavior/Features/07-NodeRemoval/01-RemoveNodeAggregate_ConstraintChecks.feature

@@ -43,7 +43,7 @@ Feature: Remove NodeAggregate
     Then the last command should have thrown an exception of type "WorkspaceDoesNotExist"
 
   Scenario: Try to remove a node aggregate in a workspace whose content stream is closed
-    When the command CloseContentStream is executed with payload:
+    When the event ContentStreamWasClosed was published with payload:
       | Key             | Value           |
       | contentStreamId | "cs-identifier" |
     When the command RemoveNodeAggregate is executed with payload and exceptions are caught:

Neos.ContentRepository.BehavioralTests/Tests/Behavior/Features/08-NodeMove/01-MoveNodes_ConstraintChecks.feature

@@ -59,7 +59,7 @@ Feature: Move node to a new parent / within the current parent before a sibling
     Then the last command should have thrown an exception of type "WorkspaceDoesNotExist"
 
   Scenario: Try to move a node in a workspace whose content stream is closed:
-    When the command CloseContentStream is executed with payload:
+    When the event ContentStreamWasClosed was published with payload:
       | Key             | Value           |
       | contentStreamId | "cs-identifier" |
     When the command MoveNodeAggregate is executed with payload and exceptions are caught: