=Still working away at the Integrations overhaul

Jake Vanderwerf

5 days ago 0dfe1d8afafc59c4a5559c498342668d5a58d6ef

 inc/managers/queue/Processor.php

@@ -14,48 +14,39 @@

   public function run(): void
   {
      if (get_transient(BASE.'queue_running')) {
         return;
      }
      set_transient(BASE.'queue_running', true, 60);
      if (!$this->hasAdequateResources()) {
         error_log('[Processor] Insufficient resources to start processing');
         return;
      }

      $ops = $this->storage->fetchRunnable();
      if (empty($ops)) {
         return;
      }
      foreach ($ops as $op) {
         if ($op->state === 'completed') {
            return;
      $op = null;
      $this->storage->withTransaction(function() use (&$op) {
         $candidates = $this->storage->fetchRunnable();
         foreach ($candidates as $candidate) {
            if ($candidate->state === 'completed') continue;
            if (!$this->dependenciesSatisfied($candidate)) continue;
            if ($this->storage->markProcessing($candidate->id)) {
               $op = $candidate;
               break;
            }
         }
         if (!$this->dependenciesSatisfied($op)) {
            continue;
         }
         if (!$this->storage->markProcessing($op->id)) {
            continue;
         }
         $this->processOne($op);
         usleep(10000);
      }
      });

      if (!$op) return;

      $this->processOne($op);
      usleep(10000);

      $this->storage->invalidateQueueCache();
   }


   private function processOne(Operation $op): void
   {
      if (get_transient(BASE.$op->id)) {
         return;
      }
      set_transient(BASE.$op->id, true, 500);
      $progress = new Progress($op);

      $executor = $this->registry->getExecutor($op->type) ?? $this->defaultExecutor;
      $op->startedAt = current_time('mysql');
      $op->state = 'processing';

      $this->storage->saveProgress($op);

      try {
@@ -227,17 +218,17 @@

      $op->errorMessage = $e->getMessage();

      JVB()->error()->log(
         '[Queue]:processOne',
         $e->getMessage(),
         [
            'operation_id' => $op->id,
            'type'         => $op->type,
            'user_id'      => $op->userId,
            'retries'      => $op->retries,
         ],
         $op->outcome === 'failed_permanent' ? 'critical' : 'warning'
      );
//    JVB()->error()->log(
//       '[Queue]:processOne',
//       $e->getMessage(),
//       [
//          'operation_id' => $op->id,
//          'type'         => $op->type,
//          'user_id'      => $op->userId,
//          'retries'      => $op->retries,
//       ],
//       $op->outcome === 'failed_permanent' ? 'critical' : 'warning'
//    );
   }

   private function calculateBackoff(int $attempt): string
@@ -288,7 +279,7 @@
      };
   }

   private function hasAdequateResources(): bool
   public function hasAdequateResources(): bool
   {
      // Stricter thresholds for starting (50% memory, 60s minimum time)
      $memoryLimit = $this->getMemoryLimitBytes();