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

[mhsdesign]

Adjustments for #5321

This pr originated after discussing NOT do do central locking #5383 and to simplify the save point creation.

A full list of discussed and here implemented changes can be seen here:
#5321 (comment)

The new catchup hook error behaviour - implemented via this pr - is documented in #4908 (comment) with great detail.

Additional technical information

---

Introduce batching in subscription engine

The replay uses a batch size of 500 (see ContentRepositoryMaintainer)
The regular catchup does not batch to ensure the catchup is run completely and any errors don't stop it
Batching does not respect the causation identifier, meaning a parent node event might be handled but not its tethered node creation event if the batch stops there - but the current 9.0 state is not different.

---

Introduce onAfterBatchCompleted hook

Previously the CatchUpHookInterface contained a onBeforeBatchCompleted which was explained as follows:

This hook is called directly before the database lock is RELEASED

It can happen that this method is called multiple times, even without having seen Events in the meantime.

If there exist more events which need to be processed, the database lock is directly acquired again after it is released.

onAfterBatchCompleted behaves similar just that it runs when the lock was released. This allows us to run all the tasks that onAfterCatchUp could also do (using possibly transactions and commiting work)

We dont ensure that onAfterBatchCompleted is only called if there are more events to handle, as this would complicate the code and technically without acquiring a lock there is never a guarantee for that, so hooks might as well encounter the case more often and have to deal with that.

---

Behaviour for catchup hooks and the same connection

Introduce test to assert behaviour when catchup hooks use the persistence manager (CatchUpHookWithPersistenceTest)

While not pretty, we rely on this working also for onAfterEvent in the redirect handler:

see https://github.com/neos/redirecthandler-neosadapter/blob/8c8f0cc6f63a030201f01cf16a7f70079149314a/Classes/Service/NodeRedirectService.php#L167

Commit the transaction during the catchup hook is evil and will not be respected. In that case the projection will be ahead of its position and an doctrine error will be thrown as no transaction is there to be committed anymore. Only a full replay can solve this if done.

We discussed that it might be sensible to use a dedicated connection for the full content repository - though for that we need some kind of abstract connection concept first.

---

Upgrade instructions

Review instructions

Checklist

• Code follows the PSR-2 coding style
• Tests have been created, run and adjusted as needed
• The PR is created against the lowest maintained branch
• Reviewer - PR Title is brief but complete and starts with FEATURE|TASK|BUGFIX
• Reviewer - The first section explains the change briefly for change-logs
• Reviewer - Breaking Changes are marked with !!! and have upgrade-instructions

[mhsdesign]

FYI, while we decided against exactly once delivery for external projections, one could still use a decorator like below.

That also allows the original test c47d181 to pass:

    /** @before */
    public function injectExternalFakeProjection(): void
    {
        $entityManager = $this->getObject(EntityManagerInterface::class);

        if (!isset(self::$secondConnection)) {
            self::$secondConnection = DriverManager::getConnection(
                $entityManager->getConnection()->getParams(),
                $entityManager->getConfiguration(),
                $entityManager->getEventManager()
            );
        }

        $this->secondFakeProjection = new DebugEventProjection(
            sprintf('cr_%s_debug_projection', self::$contentRepositoryId->value),
            self::$secondConnection
        );

        FakeProjectionFactory::setProjection(
            'second',
            ExternalDbalProjection::decorateForConnection(
                $this->secondFakeProjection,
                self::$secondConnection
            )
        );
    }

final readonly class ExternalDbalProjection implements ProjectionInterface
{
    public function __construct(
        private ProjectionInterface $decorated,
        private Connection $dbal,
    ) {
    }

    public static function decorateForConnection(
        ProjectionInterface $projection,
        Connection $dbal,
    ): self {
        return new self($projection, $dbal);
    }

    public function setUp(): void
    {
        $this->decorated->setUp();
    }

    public function status(): ProjectionStatus
    {
        return $this->decorated->status();
    }

    public function apply(EventInterface $event, EventEnvelope $eventEnvelope): void
    {
        if ($this->dbal->isTransactionActive()) {
            throw new \RuntimeException(sprintf('Expected no transaction to be active while handling decorated apply for %s', $this->decorated::class), 1733255315);
        }
        $this->dbal->transactional(fn () => $this->decorated->apply($event, $eventEnvelope));
    }

    public function getState(): ProjectionStateInterface
    {
        return $this->decorated->getState();
    }

    public function resetState(): void
    {
        $this->decorated->resetState();
    }
}

but i decided against putting that in the source code - not even for testing. Instead there is an ExternalProjectionErrorTest with the note

Note that this test only documents the current state, there were ideas to guarantee exactly once delivery for external projections
but this would mean additional complexity and is technically never truly possible with two connections.
The most likely path this might work fully is by the introduction of a decentralized subscription store concept,
see #5377, but this is out of scope for now until we find good reasons.