Implement job not found on ES retrial

webgriffe · Oct 23, 2024 · 6fbbba4 · 6fbbba4
1 parent b5dc7ae
commit 6fbbba4
Showing 1 changed file with 21 additions and 0 deletions.
diff --git a/src/Service/QueueManager.php b/src/Service/QueueManager.php
@@ -16,6 +16,9 @@
 
 final class QueueManager implements ProducerQueueManagerInterface, WorkerQueueManagerInterface
 {
+    private const JOB_NOT_FOUND_MAX_RETRIES = 3;
+    private const JOB_NOT_FOUND_RETRY_DELAY_SECONDS = 3;
+
     /**
      * @var BeanstalkClient
      */
@@ -52,6 +55,11 @@ final class QueueManager implements ProducerQueueManagerInterface, WorkerQueueMa
      */
     private $batchSize;
 
+    /**
+     * @var array<int, int>
+     */
+    private static $beanstalkIdToNotFoundCountMap = [];
+
     public function __construct(
         FlowConfig $flowConfig,
         BeanstalkClient $beanstalkClient,
@@ -140,6 +148,19 @@ public function getNextJob(): Promise
                 /** @var Job $job */
                 $job = yield $this->elasticSearch->fetchJob($jobUuid, $this->flowConfig->getTube());
             } catch (\Throwable $exception) {
+                $notFoundCount = self::$beanstalkIdToNotFoundCountMap[$jobBeanstalkId] ?? 1;
+                if ($notFoundCount <= self::JOB_NOT_FOUND_MAX_RETRIES) {
+                    self::$beanstalkIdToNotFoundCountMap[$jobBeanstalkId] = $notFoundCount + 1;
+                    $this->logger->warning(
+                        "Job with UUID {$jobUuid} not found in Elasticsearch. Retrying...",
+                        ['beanstalk_id' => $jobBeanstalkId, 'not_found_count' => $notFoundCount]
+                    );
+                    yield $this->beanstalkClient->release(
+                        $jobBeanstalkId,
+                        self::JOB_NOT_FOUND_RETRY_DELAY_SECONDS ** $notFoundCount
+                    );
+                    return yield $this->getNextJob();
+                }
                 yield $this->beanstalkClient->bury($jobBeanstalkId);
 
                 throw new JobNotFoundException($jobUuid, 0, $exception);