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