Use Scheduler as alternative to cron

refactoring command into service
usage of sync:// transport by default to preserve current behaviour

Author: rliebi

composer.json

@@ -10,17 +10,18 @@
     "psr/log": "^3.0",
     "ruflin/elastica": "^8.0.0",
     "symfony/console": "^6.2",
-    "symfony/lock": "^6.2"
+    "symfony/lock": "^6.2",
+    "symfony/scheduler": "^7.1"
   },
   "require-dev": {
     "bamarni/composer-bin-plugin": "^1.8.2",
     "phpstan/extension-installer": "^1.4.3",
-    "phpstan/phpstan": "^1.12.3",
+    "phpstan/phpstan": "^1.12.7",
     "phpstan/phpstan-deprecation-rules": "^1.2.1",
-    "phpstan/phpstan-strict-rules": "^1.6.0",
-    "rector/rector": "^1.2.5",
+    "phpstan/phpstan-strict-rules": "^1.6.1",
+    "rector/rector": "^1.2.8",
     "sentry/sentry": "^4.9.0",
-    "symfony/http-client": "^6.4.11"
+    "symfony/http-client": "^6.4.12"
   },
   "license": "MIT",
   "authors": [

src/Command/Index.php

@@ -26,7 +26,7 @@ public function getConfigTreeBuilder()
             ->integerNode('lock_timeout')->defaultValue(5 * 60)->info('To prevent overlapping indexing jobs. Set to a value higher than the slowest index. Value is specified in seconds.')->end()
             ->integerNode('cooldown')->info('Cooldown time in seconds between populating jobs. Only considered in async operation. Starts at end of last indexing job.')->defaultValue(0)->end()
             ->booleanNode('should_skip_failing_documents')->defaultFalse()->info('If true, when a document fails to be indexed, it will be skipped and indexing continue with the next document. If false, indexing that index will be aborted.')->end()
-            ->booleanNode('populate_async')->defaultFalse()->info('If true, documents to populate are being sent to the queue.')->end()
+            ->integerNode('interval')->defaultValue(600)->info('Interval in seconds at which the scheduler should be executed')->end()
             ->end()
             ->end()
             ->arrayNode('events')

src/Controller/.gitkeep

@@ -4,12 +4,15 @@
 
 namespace Valantic\ElasticaBridgeBundle\Document;
 
+use Elastic\Elasticsearch\Exception\ClientResponseException;
+use Elastic\Elasticsearch\Exception\ServerResponseException;
 use Elastica\Query\BoolQuery;
 use Elastica\Query\MatchQuery;
 use Pimcore\Model\Element\AbstractElement;
-use Valantic\ElasticaBridgeBundle\Command\Index;
+use Symfony\Contracts\Service\Attribute\Required;
 use Valantic\ElasticaBridgeBundle\Enum\DocumentType;
 use Valantic\ElasticaBridgeBundle\Index\IndexInterface;
+use Valantic\ElasticaBridgeBundle\Service\LockService;
 
 /**
  * Can be used on conjunction with DocumentNormalizerTrait::$relatedObjects.
@@ -20,20 +23,31 @@
 trait DocumentRelationAwareDataObjectTrait
 {
     protected IndexInterface $index;
+    private LockService $privateLockService;
 
     public function shouldIndex(AbstractElement $element): bool
     {
-        $result = (
-            Index::$isPopulating && $this->index->usesBlueGreenIndices()
-            ? $this->index->getBlueGreenInactiveElasticaIndex()
-            : $this->index->getElasticaIndex()
-        )
-            ->search(
-                (new BoolQuery())
-                    ->addFilter(new MatchQuery(DocumentInterface::META_TYPE, DocumentType::DOCUMENT))
-                    ->addFilter(new MatchQuery(DocumentInterface::ATTRIBUTE_RELATED_OBJECTS, $element->getId()))
-            );
+        try {
+            $result = (
+                $this->privateLockService->isPopulating($this->index) && $this->index->usesBlueGreenIndices()
+                ? $this->index->getBlueGreenInactiveElasticaIndex()
+                : $this->index->getElasticaIndex()
+            )
+                ->count(
+                    (new BoolQuery())
+                        ->addFilter(new MatchQuery(DocumentInterface::META_TYPE, DocumentType::DOCUMENT))
+                        ->addFilter(new MatchQuery(DocumentInterface::ATTRIBUTE_RELATED_OBJECTS, $element->getId())),
+                );
+        } catch (ClientResponseException|ServerResponseException) {
+            $result = 0;
+        }
 
-        return $result->count() > 0;
+        return $result > 0;
+    }
+
+    #[Required]
+    public function setLockService(LockService $lockService): void
+    {
+        $this->privateLockService = $lockService;
     }
 }

src/DependencyInjection/Configuration.php

@@ -0,0 +1,32 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Valantic\ElasticaBridgeBundle\EventListener\Messenger;
+
+use Symfony\Component\EventDispatcher\Attribute\AsEventListener;
+use Symfony\Component\EventDispatcher\EventSubscriberInterface;
+use Symfony\Component\Messenger\Event\WorkerMessageFailedEvent;
+use Valantic\ElasticaBridgeBundle\Messenger\Message\CreateDocument;
+use Valantic\ElasticaBridgeBundle\Service\LockService;
+
+#[AsEventListener]
+class CreateDocumentFailedEvent implements EventSubscriberInterface
+{
+    public function __construct(private readonly LockService $lockService) {}
+
+    public function onMessageFailed(WorkerMessageFailedEvent $event): void
+    {
+        if ($event->willRetry() || !$event->getEnvelope()->getMessage() instanceof CreateDocument) {
+            return;
+        }
+        $this->lockService->lockExecution($event->getEnvelope()->getMessage()->esIndex);
+    }
+
+    public static function getSubscribedEvents()
+    {
+        return [
+            WorkerMessageFailedEvent::class => 'onMessageFailed',
+        ];
+    }
+}

src/Document/DocumentRelationAwareDataObjectTrait.php

@@ -0,0 +1,24 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Valantic\ElasticaBridgeBundle\Exception\Index;
+
+use Valantic\ElasticaBridgeBundle\Exception\BaseException;
+
+class AlreadyInProgressException extends BaseException
+{
+    public const TYPE_COOLDOWN = 'cooldown';
+    public const TYPE_INDEXING = 'indexing';
+    public const TYPE_PROCESSING_MESSAGES = 'processing messages';
+    public const TYPE_NO_DOCUMENTS = '0 documents';
+    public const TYPE_PROCESSING = 'processing lock active';
+
+    /**
+     * @param self::TYPE_* $type
+     */
+    public function __construct(string $type)
+    {
+        parent::__construct(sprintf('Process not started (%s)', $type));
+    }
+}

src/EventListener/Messenger/CreateDocumentFailedEvent.php

@@ -58,7 +58,7 @@ final public function getCreateArguments(): array
 
     public function getBatchSize(): int
     {
-        return 5000;
+        return 500;
     }
 
     public function shouldPopulateInSubprocesses(): bool

src/Exception/Index/AlreadyInProgressException.php

@@ -117,7 +117,6 @@ public function findDocumentInstanceByPimcore(AbstractElement $element): ?Docume
      *
      * @see DocumentNormalizerTrait::$relatedObjects
      * @see IndexCommand
-     * @see IndexCommand::$isPopulating
      * @see IndexInterface::getBlueGreenInactiveElasticaIndex()
      */
     public function refreshIndexAfterEveryDocumentWhenPopulating(): bool;

src/Index/AbstractIndex.php

@@ -9,7 +9,6 @@
 use Symfony\Component\Console\Output\ConsoleOutputInterface;
 use Symfony\Component\EventDispatcher\EventDispatcherInterface;
 use Symfony\Component\Messenger\Attribute\AsMessageHandler;
-use Symfony\Component\Messenger\Exception\UnrecoverableMessageHandlingException;
 use Valantic\ElasticaBridgeBundle\Messenger\Message\CreateDocument;
 use Valantic\ElasticaBridgeBundle\Repository\ConfigurationRepository;
 use Valantic\ElasticaBridgeBundle\Repository\DocumentRepository;
@@ -30,7 +29,7 @@ public function __construct(
         private readonly ConsoleOutputInterface $consoleOutput,
     ) {}
 
-    public function __invoke(CreateDocument $message): void
+    public function __invoke(CreateDocument $message, int $retryCount = 0): void
     {
         $this->handleMessage($message);
     }
@@ -40,8 +39,9 @@ public function __invoke(CreateDocument $message): void
      * @throws ServerResponseException
      * @throws MissingParameterException
      */
-    private function handleMessage(CreateDocument $message): void
-    {
+    private function handleMessage(
+        CreateDocument $message,
+    ): void {
         $messageDecreased = false;
 
         try {
@@ -64,7 +64,7 @@ private function handleMessage(CreateDocument $message): void
                 );
             }
 
-            if ($message->callback->shouldCallEvent()) {
+            if ($message->callback?->shouldCallEvent() === true) {
                 $this->eventDispatcher->dispatch($message->callback->getEvent(), $message->callback->getEventName());
             }
 
@@ -91,27 +91,18 @@ private function handleMessage(CreateDocument $message): void
                 $this->consoleOutput->writeln(sprintf('Error processing message %s: %s', $message->esIndex, $throwable->getMessage()), ConsoleOutputInterface::VERBOSITY_NORMAL);
 
                 if (!$this->configurationRepository->shouldSkipFailingDocuments()) {
-                    $this->lockService->lockExecution($message->esIndex);
-
-                    if ($this->configurationRepository->shouldPopulateAsync()) {
-                        throw new UnrecoverableMessageHandlingException($throwable->getMessage(), previous: $throwable);
-                    }
+                    throw $throwable;
                 }
 
             }
+
             $messageDecreased = true;
             $this->lockService->messageProcessed($message->esIndex);
         } finally {
             if (!$messageDecreased) {
                 $this->consoleOutput->writeln(sprintf('Message %s not processed. (ID: %s)', $message->esIndex, $message->objectId), ConsoleOutputInterface::VERBOSITY_VERBOSE);
             }
 
-            if ($message->lastItem && $message->cooldown > 0) {
-                $key = $this->lockService->getKey($message->esIndex, 'cooldown');
-                $this->lockService->createLockFromKey($key, ttl: $message->cooldown)->acquire();
-            }
-
-
             \Pimcore::collectGarbage();
         }
     }