FEATURE: Subscription Engine

[bwaidelich]

Related: #4746

Fixes:

• Attempting to write to the content repository from two threads previously would cause likely an exception. This is more likely to happen when two importers run concurrently. A test (see ParallelWritingInWorkspaces) was added which fails in 9.0. > Failed to acquire checkpoint lock for subscriber "Neos\ContentGraph\DoctrineDbalAdapter\DoctrineDbalContentGraphProjection" because it is acquired already

TODO

• SubscriptionEngine::reset() (i.e. projection replay)
• Tests
  • spying testing catchup is correctly applied for example
  • parallel tests to ensure subscription locking works
• Fine tuning, inline docs
• Credit patchlevel (are inline doc comments for all classes underneath Neos\ContentRepository\Core\Subscription enough?)
• Optional: Allow to target specific groups/subscriptions via CLI commands (the SubscriptionEngine already supports this)

[mhsdesign]

I had a quick look and id love to help with the wiring and sticking it all together :) We should probably really get a simple draft running of the catchups just getting their projection state first :) As that will require some work already

[mhsdesign]

Okay i thought further about our little content graph projection vs projection states vs event handlers dilemma and i think the solution is not necessary exposing just the $projection on ProjectionEventHandler ... as that would also undo the explicit graph projection wiring from #5272 but instead pass a "bigger" object than "just" the Subscribers into the content repository factory.

in my eyes this object, being build by the content repository registry in some way, must reflect

• what is guaranteed to be the ContentGraphProjectionInterface and its state ContentGraphReadModelInterface, not part of some collection but distinct
• what are the additional projection states
• what are all the subscribers (with their catchup-hooks) that are just directly passed to the subscription engine without doing anything further with it. (e.g. no public accessors on the ProjectionEventHandler for anything)

that could look like:

final readonly class ContentRepositoryGraphProjectionAndSubscribers
{
    public function __construct(
        public ContentGraphProjectionInterface $contentGraphProjection,
        public Subscribers $subscribers, // must contain a subscriber for the $contentGraphProjection
        public ProjectionStates $additionalProjectionStates, // must not contain the $contentGraphProjection state
    ) {
    }
}

or maybe a little more explicit so the factories dont have to deal with all the logic and we have control over the subscription ids:

final readonly class ProjectionsAndCatchupHooksBetterVersionZwo
{
    public function __construct(
        public ContentGraphProjectionInterface $contentGraphProjection,
        private Projections $additionalProjections,
        private Subscribers $additionalSubscriber,
        private array $catchUpHooksByProjectionClass
    ) {
    }

    public function getSubscribers(): Subscribers
    {
        $subscribers = iterator_to_array($this->additionalSubscriber);

        $subscribers[] = new Subscriber(
            SubscriptionId::fromString('contentGraphProjection'),
            SubscriptionGroup::fromString('default'),
            RunMode::FROM_BEGINNING,
            new ProjectionEventHandler(
                $this->contentGraphProjection,
                $this->getCatchUpHooksForProjectionClass(ContentGraphProjectionInterface::class),
            ),
        );
        
        foreach ($this->additionalProjections as $projection) {
            $subscribers[] = new Subscriber(
                SubscriptionId::fromString(substr(strrchr($projection::class, '\\'), 1)),
                SubscriptionGroup::fromString('default'),
                RunMode::FROM_BEGINNING,
                new ProjectionEventHandler(
                    $projection,
                    $this->getCatchUpHooksForProjectionClass($projection::class),
                ),
            );
        }
        
        return Subscribers::fromArray($subscribers);
    }
    
    public function getAdditionalProjectionStates(): ProjectionStates
    {
        return ProjectionStates::fromArray(array_map(
            fn ($projection) => $projection->getState(),
            iterator_to_array($this->additionalProjections)
        ));
    }

    private function getCatchUpHooksForProjectionClass(string $projectionClass): ?CatchUpHookInterface
    {
        return $this->catchUpHooksByProjectionClass[$projectionClass] ?? null;
    }
}

but for things that will belong to the future ProjectionService like, replayProjection, replayAllProjections, resetAllProjections we might still need to expose all projections here, unless the subscription engine will lear that itself: $this->subscriptionEngine->reset()

[bwaidelich]

@mhsdesign thanks for your input!
Ands in a class name always make me suspicious..

That's my current draft of the ContentRepositoryFactory constructor:

public function __construct(
    private readonly ContentRepositoryId $contentRepositoryId,
    EventStoreInterface $eventStore,
    NodeTypeManager $nodeTypeManager,
    ContentDimensionSourceInterface $contentDimensionSource,
    Serializer $propertySerializer,
    private readonly UserIdProviderInterface $userIdProvider,
    private readonly ClockInterface $clock,
    SubscriptionStoreInterface $subscriptionStore,
    ContentGraphProjectionFactoryInterface $contentGraphProjectionFactory,
    CatchUpHookFactoryInterface $contentGraphCatchUpHookFactory,
    private readonly ContentRepositorySubscribersFactoryInterface $additionalSubscribersFactory,
) {
// ...
}

[mhsdesign]

As discussed that looks good ❤️ my idea had a flaw because i assumed the projection instance could be build by the registry which it CANNOT because we need factory dependencies.... and the thing with iterating over the event handlers to fetch their state via getState or something is weird but okay in as that projectionState is now a little tunnel through space as well :) So definitely okay to do that little quirk.

[mhsdesign]

a little update from my part:

i spend a lot of time first getting a little bit of testcoverage before i trusted myself to do any core changes :D
So right now i have everything i want i think including a spying fake projection to check which events were applied and assert events are only applied once, as well as the capabilities to check how catchups work and especially how does everything behave in case of a projection or catchup error.

• Test for CatchUpHookError
• Test for CatchUpHook
• Test for ProjectionError
• Test for SubscriptionActiveStatus
• Test for SubscriptionBootingStatus
• Test for SubscriptionDetachedStatus
• Test for SubscriptionGetStatus
• Test for SubscriptionNewStatus
• Test for SubscriptionReset
• Test for SubscriptionSetup

bugs i discovered and fixed for the tests:

• BUGFIX: subscriptionStatuses should work if no schema
• BUGFIX: Detached subscriber status throws because projection status cannot be read
• BUGFIX: reset error subscription state on reset
• BUGFIX: Reintroduce catchup hooks for additional projections

---

Deliberate behaviour changes / simplifications:

• TASK: Inline discoverDetachedSubscriptions
• TASK: Do not discover new subscriptions during catchup
• TASK: Remove retry_attempt from subscriptions and everything else from the cleanup: TASK: Radical cleanup for subscription pr to simplify and get into 9.0 #5375

---

introduction of savepoints

The previous checkpoint implementation guaranteed an exactly once delivery. (implemented by using a transaction for each event for each projection in a for update lock)

We still need this rollback behaviour: If a catchup fails on the content graph projection, it means the content graph projection is in error state but has the event applied and would get it after repairing the catchup again, that will not work for the projection.
And the rollback also greatly simplifies to reason about the catchup hooks and if they have to deal with an at least once delivery or now.
Also the case where a projection is really broken it will not be trivial to fix if the state is possibly already partially messed up.

Currently the subscription engine uses a big transaction for all projections and to update the subscriptions at the end. We cannot roll that back. So we want the following things:

• a for update lock in ONE transaction (allows to terminate the cli process via CTRL+C without any locking artefacts)
• fetching all subscribers at once instead of one by one
• adding an error state if the projection failed and rollback, but keep not failed events on that projection

i implemented this with savepoints which would look like:

BEGIN
SELECT subscriptions FOR UPDATE ...

SAVEPOINT CATCHUP
INSERT INTO projection_A (event1)
RELEASE SAVEPOINT CATCHUP

UPDATE subscriptions ...
COMMIT

the exactly once delivery is well tested with my above tests.

and looks like this on error:

BEGIN
SELECT subscriptions FOR UPDATE ...

SAVEPOINT CATCHUP
INSERT INTO projection_A (event1)
ROLLBACK TO SAVEPOINT CATCHUP

UPDATE subscriptions ... // update the error state
COMMIT

Regarding onBeforeCatchUp and onAfterCatchUp, we cannot wrap these into a savepoint but will just throw hard and consider this a not forgive-able developer error. The whole transaction will be rolled back.: TASK: Throw CatchUpFailed exception in case onBeforeCatchUp or onAfterCatchUp failes

followup: inconsistency of projection transaction vs subscription transaction.

In this commit TASK: Use save points to rollback projections during transaction on failure i added methods on the subscription store to set the savepoints. This is not entirely correct as there is no rule that a projection and its store have to share the same dbal instance or even connection nor database.

In the current state of 9.0 this is solved by using the pattern of a checkpoint table per projection, which would live in the same database and transaction.
One central store cannot offer that flexibility. In case we want to support that each projection can live in their own database we probably need a solution like this to attach the subscription store to the subscriber: #5377 and on catchup iterate over all stores (deduplicated) and their subscribers. That works but brings some new complexity for wiring and also for the detached state.

Instead for now it might be sensible to add a transactional function to the projection where we use savepoints to emulate nested transactions. In case that no transaction is open, we can throw an exception stating that its not possible yet to decouple projection from store:

public function transactional(\Closure $fn): mixed
{
    if ($this->dbal->isTransactionActive() === false) {
        throw new \RuntimeException(sprintf('Cannot set savepoint for projection, because no transaction is active. SubscriptionEngine and Projection must use the same database connection.'));
    }
    $this->dbal->createSavepoint('PROJECTION');
    try {
        $result = $fn();
    } catch (\Throwable $e) {
        $this->dbal->rollbackSavepoint('PROJECTION');
        throw $e;
    }
    $this->dbal->releaseSavepoint('PROJECTION');
    return $result;
}

---

mutable SubscriptionManager

regarding the mutable vs immutable subscription updating, i have a wip local which does simplify this.

I'm happy if you can simplify this ofc but it's important that the subscription states only get persisted once just befor the transaction is comitted.

as far as i can see its only important to reduce sql statements, but for the result it should not matter. In my case, marking the projection as failure happens directly, while the other active and position updates happen at the end.

[mhsdesign]

@kitsunet and me agreed on the following:

In order to get the basis of the change in we should keep the locking behaviour as it is now (meaning one transaction for each event for each projection) as this will allow discussing behaviour changes in a separate pull request and also if a simplification to reduce transactions brings really any performance. So this pr would only introduce the central registration (which allows to fetch the lowest position in one query) and leave us with the performance gain of only having to load the events one and not n times on replay.

That means each projection will receive a function which either uses save points to emulate nested transactions, or if its another connection or database uses an own transaction:

public function transactional(\Closure $fn): mixed
{
    if ($this->dbal->isTransactionActive() === false) {
        return $this->dbal->transactional($fn);
    }
    $this->dbal->createSavepoint('PROJECTION');
    try {
        $result = $fn();
    } catch (\Throwable $e) {
        $this->dbal->rollbackSavepoint('PROJECTION');
        throw $e;
    }
    $this->dbal->releaseSavepoint('PROJECTION');
    return $result;
}

Our reasons for still replying so hard on exactly once delivery are that our projections simply are not build to ignore events twice. Until we fully agree that this is a good idea, and also find solutions to do the same for catchup hooks, and have test that prove that at least once delivery is okay in most cases, it would be not a good idea to introduce a change here before done the other.

future followups

• experiment with a write table lock (which is on connection basis, meaning it should be unlocked if the process crashes) that would allow us to: Lock, fetch lowest position, open transaction (without for update) and commit.
• rediscuss and test if only using one transaction is dramatically faster which could lead us to decide to not support projections in other databases
• rethink FOR UPDATE NOAWAIT locking, as two importers would shoot each other down: will be part of WIP: Remove NOWAIT? #5376