Implement CatchUpDeduplication with locks

Using symfony/lock ensures atomic locking, which should really prevent duplication even under load.
neos · Dec 6, 2023 · 5299fcc · 5299fcc
1 parent 5e31584
commit 5299fcc
Show file tree

Hide file tree

Showing 5 changed files with 49 additions and 62 deletions.
diff --git a/Neos.ContentRepositoryRegistry/Classes/ContentRepositoryRegistry.php b/Neos.ContentRepositoryRegistry/Classes/ContentRepositoryRegistry.php
@@ -32,6 +32,9 @@
 use Neos\Utility\Arrays;
 use Neos\Utility\PositionalArraySorter;
 use Psr\Clock\ClockInterface;
+use Symfony\Component\Lock\LockFactory;
+use Symfony\Component\Lock\PersistingStoreInterface;
+use Symfony\Component\Lock\Store\DoctrineDbalStore;
 use Symfony\Component\Serializer\Normalizer\DenormalizerInterface;
 use Symfony\Component\Serializer\Normalizer\NormalizerInterface;
 use Symfony\Component\Serializer\Serializer;
@@ -238,14 +241,20 @@ private function buildCatchUpDeduplicationQueue(ContentRepositoryId $contentRepo
         }
         $projectionCatchUpTrigger = $projectionCatchUpTriggerFactory->build($contentRepositoryId, $contentRepositorySettings['projectionCatchUpTrigger']['options'] ?? []);
 
-        $catchUpStateCache = $this->objectManager->get('Neos.ContentRepositoryRegistry:CacheCatchUpStates');
-        if (!$catchUpStateCache instanceof FrontendInterface) {
-            throw InvalidConfigurationException::fromMessage('The virtual object "Neos.ContentRepositoryRegistry:CacheCatchUpStates" must provide a Cache Frontend, but is "%s".', get_debug_type($catchUpStateCache));
+        $catchUpStateLockStorage = $this->objectManager->get('Neos.ContentRepositoryRegistry:QueueLockStorage');
+        if (!$catchUpStateLockStorage instanceof PersistingStoreInterface) {
+            throw InvalidConfigurationException::fromMessage('The virtual object "Neos.ContentRepositoryRegistry:QueueLockStorage" must provide a \Symfony\Component\Lock\PersistingStoreInterface, but is "%s".', get_debug_type($catchUpStateLockStorage));
+        }
+        if ($catchUpStateLockStorage instanceof DoctrineDbalStore) {
+            try {
+                // hack to ensure tables exist for Dbal
+                $catchUpStateLockStorage->createTable();
+            } catch (\Doctrine\DBAL\Exception\TableExistsException $_) {}
         }
 
         return new CatchUpDeduplicationQueue(
             $contentRepositoryId,
-            $catchUpStateCache,
+            new LockFactory($catchUpStateLockStorage),
             $projectionCatchUpTrigger
         );
     }

diff --git a/Neos.ContentRepositoryRegistry/Classes/Service/CatchUpDeduplicationQueue.php b/Neos.ContentRepositoryRegistry/Classes/Service/CatchUpDeduplicationQueue.php
@@ -1,10 +1,10 @@
 <?php
 namespace Neos\ContentRepositoryRegistry\Service;
 
-use Neos\Cache\Frontend\FrontendInterface;
 use Neos\ContentRepository\Core\Factory\ContentRepositoryId;
 use Neos\ContentRepository\Core\Projection\ProjectionCatchUpTriggerInterface;
 use Neos\ContentRepository\Core\Projection\Projections;
+use Symfony\Component\Lock\LockFactory;
 
 /**
  * This encapsulates logic to provide exactly once catchUps
@@ -15,7 +15,7 @@
 {
     public function __construct(
         private ContentRepositoryId $contentRepositoryId,
-        private FrontendInterface $catchUpLock,
+        private LockFactory $lockFactoy,
         private ProjectionCatchUpTriggerInterface $projectionCatchUpTrigger
     ) {}
 
@@ -46,24 +46,25 @@ public function requestCatchUp(Projections $projections): void
      */
     public function releaseCatchUpLock(string $projectionClassName): void
     {
-        $this->catchUpLock->remove($this->cacheKeyRunning($projectionClassName));
+        $runningLock = $this->lockFactoy->createLock($this->cacheKeyRunning($projectionClassName));
+        $runningLock->isAcquired() && $runningLock->release();
     }
 
     private function triggerCatchUpAndReturnQueued(Projections $projections): Projections
     {
         $projectionsToCatchUp = [];
         $queuedProjections = [];
         foreach ($projections as $projection) {
-            if (!$this->isRunning($projection::class)) {
-                $this->run($projection::class);
+            $runningLock = $this->lockFactoy->createLock($this->cacheKeyRunning($projection::class));
+            $queuedLock = $this->lockFactoy->createLock($this->cacheKeyQueued($projection::class));
+            if ($runningLock->acquire()) {
                 // We are about to start a catchUp and can therefore discard any queue that exists right now, apparently someone else is waiting for it.
-                $this->dequeue($projection::class);
+                $queuedLock->release();
                 $projectionsToCatchUp[] = $projection;
                 continue;
             }
 
-            if (!$this->isQueued($projection::class)) {
-                $this->queue($projection::class);
+            if ($queuedLock->acquire()) {
                 $queuedProjections[] = $projection;
             }
         }
@@ -77,16 +78,19 @@ private function retryQueued(Projections $queuedProjections): Projections
     {
         $passToCatchUp = [];
         $stillQueuedProjections = [];
+
         foreach ($queuedProjections as $projection) {
-            if (!$this->isQueued($projection::class)) {
+            $runningLock = $this->lockFactoy->createLock($this->cacheKeyRunning($projection::class));
+            $queuedLock = $this->lockFactoy->createLock($this->cacheKeyQueued($projection::class));
+
+            if (!$queuedLock->isAcquired()) {
                 // was dequeued, we can drop it
                 continue;
             }
 
-            if (!$this->isRunning($projection::class)) {
-                $this->run($projection::class);
+            if ($runningLock->acquire()) {
                 // We are about to start a catchUp and can therefore discard any queue that exists right now, apparently someone else is waiting for it.
-                $this->dequeue($projection::class);
+                $queuedLock->release();
                 $passToCatchUp[] = $projection;
                 continue;
             }
@@ -99,51 +103,6 @@ private function retryQueued(Projections $queuedProjections): Projections
         return Projections::fromArray($stillQueuedProjections);
     }
 
-    /**
-     * @param class-string $projectionClassName
-     * @return bool
-     */
-    private function isRunning(string $projectionClassName): bool
-    {
-        return $this->catchUpLock->has($this->cacheKeyRunning($projectionClassName));
-    }
-
-    /**
-     * @param class-string $projectionClassName
-     * @return void
-     */
-    private function run(string $projectionClassName): void
-    {
-        $this->catchUpLock->set($this->cacheKeyRunning($projectionClassName), 1);
-    }
-
-    /**
-     * @param class-string $projectionClassName
-     * @return bool
-     */
-    private function isQueued(string $projectionClassName): bool
-    {
-        return $this->catchUpLock->has($this->cacheKeyQueued($projectionClassName));
-    }
-
-    /**
-     * @param class-string $projectionClassName
-     * @return void
-     */
-    private function queue(string $projectionClassName): void
-    {
-        $this->catchUpLock->set($this->cacheKeyQueued($projectionClassName), 1);
-    }
-
-    /**
-     * @param class-string $projectionClassName
-     * @return void
-     */
-    private function dequeue(string $projectionClassName): void
-    {
-        $this->catchUpLock->remove($this->cacheKeyQueued($projectionClassName));
-    }
-
     /**
      * @param class-string $projectionClassName
      * @return string

diff --git a/Neos.ContentRepositoryRegistry/Configuration/Objects.yaml b/Neos.ContentRepositoryRegistry/Configuration/Objects.yaml
@@ -41,3 +41,20 @@ Neos\ContentRepository\Core\Projection\NodeHiddenState\NodeHiddenStateProjection
   arguments:
     1:
       value: Neos_ContentRepositoryRegistry_CatchUpStates
+
+'Neos.ContentRepositoryRegistry:DbalConnection':
+  className: Doctrine\DBAL\Connection
+  factoryObjectName: Neos\ContentRepositoryRegistry\DoctrineDbalClient\DoctrineDbalClient
+  factoryMethodName: getConnection
+
+'Neos.ContentRepositoryRegistry:QueueLockStorage':
+  className: Symfony\Component\Lock\Store\DoctrineDbalStore
+  arguments:
+    1:
+      object: 'Neos.ContentRepositoryRegistry:DbalConnection'
+
+'Neos.ContentRepositoryRegistry:QueueLockFactory':
+  className: Symfony\Component\Lock\LockFactory
+  arguments:
+    1:
+      object: 'Neos.ContentRepositoryRegistry:QueueLockStorage'
diff --git a/Neos.ContentRepositoryRegistry/composer.json b/Neos.ContentRepositoryRegistry/composer.json
@@ -17,7 +17,8 @@
         "neos/contentrepository-core": "self.version",
         "neos/contentrepositoryregistry-storageclient": "self.version",
         "symfony/property-access": "^5.4|^6.0",
-        "psr/clock": "^1"
+        "psr/clock": "^1",
+        "symfony/lock": "^6.0.0"
     },
     "autoload": {
         "psr-4": {

diff --git a/composer.json b/composer.json
@@ -26,6 +26,7 @@
         "neos/behat": "^9.0",
         "neos/contentrepositoryregistry-storageclient": "self.version",
         "symfony/property-access": "^5.4|^6.0",
+        "symfony/lock": "^6.0.0",
         "neos/fluid-adaptor": "*",
         "neos/cache": "*",
         "neos/eel": "*",