BUGFIX: Ensure users content stream is never left closed after publication

Neos.ContentRepository.Core/Classes/CommandHandler/CommandBus.php

@@ -11,6 +11,7 @@
  * Implementation Detail of {@see ContentRepository::handle}, which does the command dispatching to the different
  * {@see CommandHandlerInterface} implementation.
  *
+ * @phpstan-import-type YieldedEventsToPublish from CommandHandlerInterface
  * @internal
  */
 final readonly class CommandBus
@@ -29,7 +30,10 @@ public function __construct(
     }
 
     /**
-     * @return EventsToPublish|\Generator<int, EventsToPublish>
+     * The handler only calculate which events they want to have published,
+     * but do not do the publishing themselves
+     *
+     * @return EventsToPublish|YieldedEventsToPublish
      */
     public function handle(CommandInterface $command): EventsToPublish|\Generator
     {

Neos.ContentRepository.Core/Classes/CommandHandler/CommandHandlerInterface.php

@@ -5,12 +5,14 @@
 namespace Neos\ContentRepository\Core\CommandHandler;
 
 use Neos\ContentRepository\Core\EventStore\EventsToPublish;
+use Neos\EventStore\Model\EventStore\CommitResult;
 
 /**
  * Common interface for all Content Repository command handlers
  *
  * The {@see CommandHandlingDependencies} are available during handling to do soft-constraint checks
  *
+ * @phpstan-type YieldedEventsToPublish \Generator<int, EventsToPublish, CommitResult|null, void>
  * @internal no public API, because commands are no extension points of the CR
  */
 interface CommandHandlerInterface
@@ -23,7 +25,7 @@ public function canHandle(CommandInterface $command): bool;
      * For the case of the workspace command handler that need to publish to many streams and "close" the content-stream directly,
      * it's allowed to yield the events to interact with the control flow of event publishing.
      *
-     * @return EventsToPublish|\Generator<int, EventsToPublish>
+     * @return EventsToPublish|YieldedEventsToPublish
      */
     public function handle(CommandInterface $command, CommandHandlingDependencies $commandHandlingDependencies): EventsToPublish|\Generator;
 }

Neos.ContentRepository.Core/Classes/ContentRepository.php

@@ -94,19 +94,35 @@ public function __construct(
      */
     public function handle(CommandInterface $command): void
     {
-        // the commands only calculate which events they want to have published, but do not do the
-        // publishing themselves
-        $eventsToPublishOrGenerator = $this->commandBus->handle($command);
-
-        if ($eventsToPublishOrGenerator instanceof EventsToPublish) {
-            $eventsToPublish = $this->enrichEventsToPublishWithMetadata($eventsToPublishOrGenerator);
-            $this->eventPersister->publishEvents($this, $eventsToPublish);
-        } else {
-            foreach ($eventsToPublishOrGenerator as $eventsToPublish) {
-                assert($eventsToPublish instanceof EventsToPublish); // just for the ide
-                $eventsToPublish = $this->enrichEventsToPublishWithMetadata($eventsToPublish);
-                $this->eventPersister->publishEvents($this, $eventsToPublish);
+        $toPublish = $this->commandBus->handle($command);
+
+        if ($toPublish instanceof EventsToPublish) {
+            // simple case
+            $eventsToPublish = $this->enrichEventsToPublishWithMetadata($toPublish);
+            if ($eventsToPublish->events->isEmpty()) {
+                return;
+            }
+            $this->eventPersister->publishWithoutCatchup($eventsToPublish);
+            $this->catchupProjections();
+            return;
+        }
+
+        // control-flow aware command handling via generator
+        try {
+            $yieldedEventsToPublish = $toPublish->current();
+            while ($yieldedEventsToPublish !== null) {
+                if ($yieldedEventsToPublish->events->isEmpty()) {
+                    $yieldedEventsToPublish = $toPublish->send(null);
+                    continue;
+                }
+                $eventsToPublish = $this->enrichEventsToPublishWithMetadata($yieldedEventsToPublish);
+                $commitResult = $this->eventPersister->publishWithoutCatchup($eventsToPublish);
+                $yieldedEventsToPublish = $toPublish->send($commitResult);
             }
+        } finally {
+            // We always NEED to catchup even if there was an unexpected ConcurrencyException to make sure previous commits are handled.
+            // Technically it would be acceptable for the catchup to fail here (due to hook errors) because all the events are already persisted.
+            $this->catchupProjections();
         }
     }
 

Neos.ContentRepository.Core/Classes/EventStore/EventPersister.php

@@ -8,6 +8,7 @@
 use Neos\EventStore\EventStoreInterface;
 use Neos\EventStore\Exception\ConcurrencyException;
 use Neos\EventStore\Model\Events;
+use Neos\EventStore\Model\EventStore\CommitResult;
 
 /**
  * Internal service to persist {@see EventInterface} with the proper normalization, and triggering the
@@ -31,15 +32,22 @@ public function publishEvents(ContentRepository $contentRepository, EventsToPubl
         if ($eventsToPublish->events->isEmpty()) {
             return;
         }
+        $this->publishWithoutCatchup($eventsToPublish);
+        $contentRepository->catchUpProjections();
+    }
+
+    /**
+     * @throws ConcurrencyException in case the expectedVersion does not match
+     */
+    public function publishWithoutCatchup(EventsToPublish $eventsToPublish): CommitResult
+    {
         $normalizedEvents = Events::fromArray(
             $eventsToPublish->events->map($this->eventNormalizer->normalize(...))
         );
-        $this->eventStore->commit(
+        return $this->eventStore->commit(
             $eventsToPublish->streamName,
             $normalizedEvents,
             $eventsToPublish->expectedVersion
         );
-
-        $contentRepository->catchUpProjections();
     }
 }

Neos.ContentRepository.Core/Classes/Feature/ContentStreamHandling.php

@@ -9,55 +9,25 @@
 use Neos\ContentRepository\Core\EventStore\EventsToPublish;
 use Neos\ContentRepository\Core\Feature\ContentStreamClosing\Event\ContentStreamWasClosed;
 use Neos\ContentRepository\Core\Feature\ContentStreamClosing\Event\ContentStreamWasReopened;
-use Neos\ContentRepository\Core\Feature\ContentStreamCreation\Event\ContentStreamWasCreated;
 use Neos\ContentRepository\Core\Feature\ContentStreamForking\Event\ContentStreamWasForked;
 use Neos\ContentRepository\Core\Feature\ContentStreamRemoval\Event\ContentStreamWasRemoved;
 use Neos\ContentRepository\Core\SharedModel\Exception\ContentStreamAlreadyExists;
 use Neos\ContentRepository\Core\SharedModel\Exception\ContentStreamDoesNotExistYet;
 use Neos\ContentRepository\Core\SharedModel\Exception\ContentStreamIsClosed;
-use Neos\ContentRepository\Core\SharedModel\Exception\ContentStreamIsNotClosed;
 use Neos\ContentRepository\Core\SharedModel\Workspace\ContentStreamId;
+use Neos\EventStore\Model\Event\Version;
 use Neos\EventStore\Model\EventStream\ExpectedVersion;
 
 trait ContentStreamHandling
 {
-    /**
-     * @param ContentStreamId $contentStreamId The id of the content stream to create
-     * @throws ContentStreamAlreadyExists
-     * @phpstan-pure this method is pure, to persist the events they must be handled outside
-     */
-    private function createContentStream(
-        ContentStreamId $contentStreamId,
-        CommandHandlingDependencies $commandHandlingDependencies,
-    ): EventsToPublish {
-        $this->requireContentStreamToNotExistYet($contentStreamId, $commandHandlingDependencies);
-        $streamName = ContentStreamEventStreamName::fromContentStreamId($contentStreamId)
-            ->getEventStreamName();
-
-        return new EventsToPublish(
-            $streamName,
-            Events::with(
-                new ContentStreamWasCreated(
-                    $contentStreamId,
-                )
-            ),
-            ExpectedVersion::NO_STREAM()
-        );
-    }
-
     /**
      * @param ContentStreamId $contentStreamId The id of the content stream to close
-     * @param CommandHandlingDependencies $commandHandlingDependencies
-     * @return EventsToPublish
      * @phpstan-pure this method is pure, to persist the events they must be handled outside
      */
     private function closeContentStream(
         ContentStreamId $contentStreamId,
-        CommandHandlingDependencies $commandHandlingDependencies,
+        Version $contentStreamVersion,
     ): EventsToPublish {
-        $this->requireContentStreamToExist($contentStreamId, $commandHandlingDependencies);
-        $expectedVersion = $this->getExpectedVersionOfContentStream($contentStreamId, $commandHandlingDependencies);
-        $this->requireContentStreamToNotBeClosed($contentStreamId, $commandHandlingDependencies);
         $streamName = ContentStreamEventStreamName::fromContentStreamId($contentStreamId)->getEventStreamName();
 
         return new EventsToPublish(
@@ -67,29 +37,26 @@ private function closeContentStream(
                     $contentStreamId,
                 ),
             ),
-            $expectedVersion
+            ExpectedVersion::fromVersion($contentStreamVersion)
         );
     }
 
     /**
      * @param ContentStreamId $contentStreamId The id of the content stream to reopen
      * @phpstan-pure this method is pure, to persist the events they must be handled outside
      */
-    private function reopenContentStream(
+    private function reopenContentStreamWithoutConstraints(
         ContentStreamId $contentStreamId,
-        CommandHandlingDependencies $commandHandlingDependencies,
     ): EventsToPublish {
-        $this->requireContentStreamToExist($contentStreamId, $commandHandlingDependencies);
-        $this->requireContentStreamToBeClosed($contentStreamId, $commandHandlingDependencies);
-        $streamName = ContentStreamEventStreamName::fromContentStreamId($contentStreamId)->getEventStreamName();
-
         return new EventsToPublish(
-            $streamName,
+            ContentStreamEventStreamName::fromContentStreamId($contentStreamId)->getEventStreamName(),
             Events::with(
                 new ContentStreamWasReopened(
                     $contentStreamId
                 ),
             ),
+            // We operate here without constraints on purpose to ensure this can be commited.
+            //Constraints have been checked beforehand and its expected that the content stream is closed.
             ExpectedVersion::ANY()
         );
     }
@@ -104,19 +71,10 @@ private function reopenContentStream(
     private function forkContentStream(
         ContentStreamId $newContentStreamId,
         ContentStreamId $sourceContentStreamId,
-        CommandHandlingDependencies $commandHandlingDependencies
+        Version $sourceContentStreamVersion
     ): EventsToPublish {
-        $this->requireContentStreamToExist($sourceContentStreamId, $commandHandlingDependencies);
-        $this->requireContentStreamToNotBeClosed($sourceContentStreamId, $commandHandlingDependencies);
-        $this->requireContentStreamToNotExistYet($newContentStreamId, $commandHandlingDependencies);
-
-        $sourceContentStreamVersion = $commandHandlingDependencies->getContentStreamVersion($sourceContentStreamId);
-
-        $streamName = ContentStreamEventStreamName::fromContentStreamId($newContentStreamId)
-            ->getEventStreamName();
-
         return new EventsToPublish(
-            $streamName,
+            ContentStreamEventStreamName::fromContentStreamId($newContentStreamId)->getEventStreamName(),
             Events::with(
                 new ContentStreamWasForked(
                     $newContentStreamId,
@@ -133,25 +91,19 @@ private function forkContentStream(
      * @param ContentStreamId $contentStreamId The id of the content stream to remove
      * @phpstan-pure this method is pure, to persist the events they must be handled outside
      */
-    private function removeContentStream(
+    private function removeContentStreamWithoutConstraints(
         ContentStreamId $contentStreamId,
-        CommandHandlingDependencies $commandHandlingDependencies
     ): EventsToPublish {
-        $this->requireContentStreamToExist($contentStreamId, $commandHandlingDependencies);
-        $expectedVersion = $this->getExpectedVersionOfContentStream($contentStreamId, $commandHandlingDependencies);
-
-        $streamName = ContentStreamEventStreamName::fromContentStreamId(
-            $contentStreamId
-        )->getEventStreamName();
-
         return new EventsToPublish(
-            $streamName,
+            ContentStreamEventStreamName::fromContentStreamId($contentStreamId)->getEventStreamName(),
             Events::with(
                 new ContentStreamWasRemoved(
                     $contentStreamId,
                 ),
             ),
-            $expectedVersion
+            // We operate here without constraints on purpose to ensure this can be commited.
+            // Constraints have been checked beforehand and its expected that the content stream is closed.
+            ExpectedVersion::ANY()
         );
     }
 
@@ -172,23 +124,6 @@ private function requireContentStreamToNotExistYet(
         }
     }
 
-    /**
-     * @param ContentStreamId $contentStreamId
-     * @param CommandHandlingDependencies $commandHandlingDependencies
-     * @throws ContentStreamDoesNotExistYet
-     */
-    private function requireContentStreamToExist(
-        ContentStreamId $contentStreamId,
-        CommandHandlingDependencies $commandHandlingDependencies
-    ): void {
-        if (!$commandHandlingDependencies->contentStreamExists($contentStreamId)) {
-            throw new ContentStreamDoesNotExistYet(
-                'Content stream "' . $contentStreamId->value . '" does not exist yet.',
-                1521386692
-            );
-        }
-    }
-
     private function requireContentStreamToNotBeClosed(
         ContentStreamId $contentStreamId,
         CommandHandlingDependencies $commandHandlingDependencies
@@ -200,24 +135,4 @@ private function requireContentStreamToNotBeClosed(
             );
         }
     }
-
-    private function requireContentStreamToBeClosed(
-        ContentStreamId $contentStreamId,
-        CommandHandlingDependencies $commandHandlingDependencies
-    ): void {
-        if (!$commandHandlingDependencies->isContentStreamClosed($contentStreamId)) {
-            throw new ContentStreamIsNotClosed(
-                'Content stream "' . $contentStreamId->value . '" is not closed.',
-                1710405911
-            );
-        }
-    }
-
-    private function getExpectedVersionOfContentStream(
-        ContentStreamId $contentStreamId,
-        CommandHandlingDependencies $commandHandlingDependencies
-    ): ExpectedVersion {
-        $version = $commandHandlingDependencies->getContentStreamVersion($contentStreamId);
-        return ExpectedVersion::fromVersion($version);
-    }
 }