MikoPBX\PBXCoreREST\Controllers\BaseController::sendRequestToBackendWorker()

Complexity

Conditions	27
Paths	17058

Size

Total Lines	347
Code Lines	199

Duplication

Lines	0
Ratio	0 %

Importance

Changes	1
Bugs	0	Features	0

<?php
    public function sendRequestToBackendWorker(
        string $processor,
        string $actionName,
        mixed $payload = null,
        string $moduleName = '',
        int $maxTimeout = 30,
        int $priority = 0

The parameter $priority is not used and could be removed.

    ): void {
        [$debug, $requestMessage] = $this->prepareRequestMessage($processor, $payload, $actionName, $moduleName);
        if ($debug) {
            $maxTimeout = 9999;
        } else {
            // Ensure minimum timeout for testing
            $maxTimeout = max($maxTimeout, 30);
        }

        try {
            // Initialize Redis connection
            $redis = $this->di->get(RedisClientProvider::SERVICE_NAME);

            // Generate unique request ID
            $requestMessage['request_id'] = uniqid("req_{$actionName}_", true);

            // Push request to queue
            $pushResult = $redis->rpush(WorkerApiCommands::REDIS_API_QUEUE, json_encode($requestMessage));
            
            // Проверяем, что задача действительно добавлена в очередь
            if ($pushResult <= 0) {
                throw new \RuntimeException("Failed to push request to Redis queue");
            }
            
            // Проверяем текущий размер очереди и логируем
            $queueLength = $redis->lLen(WorkerApiCommands::REDIS_API_QUEUE);
            SystemMessages::sysLogMsg(
                static::class,
                sprintf(
                    "Request added to queue: action=%s, id=%s, queue_position=%d/%d",
                    $actionName,
                    $requestMessage['request_id'],
                    $pushResult,
                    $queueLength
                ),
                LOG_DEBUG
            );
            
            // Подсчитываем, сколько активных воркеров доступны для обработки
            $activeWorkers = $redis->keys('worker_api_commands:*');
            $runningWorkers = count($activeWorkers);
            
            // Если в очереди много запросов, но мало воркеров, логируем предупреждение
            if ($queueLength > $runningWorkers * 2) {
                SystemMessages::sysLogMsg(
                    static::class,
                    sprintf(
                        "WARNING: Queue backlog detected - %d requests in queue with only %d workers",
                        $queueLength,
                        $runningWorkers
                    ),
                    LOG_WARNING
                );
            }

            if ($requestMessage['async']) {
                $this->response->setPayloadSuccess(['success' => true]);
                return;
            }

            // Get response key
            $responseKey = WorkerApiCommands::REDIS_API_RESPONSE_PREFIX . $requestMessage['request_id'];
            $response = null;
            $startTime = microtime(true);
            $retryCount = 0;
            $maxRetries = 5;
            
            // Calculate end time
            $endTime = $startTime + $maxTimeout;
            
            // Polling intervals strategy: fast polling at first, then gradually slow down
            $pollingIntervals = [
                // Fast polling for first second (10ms)
                ['duration' => 1, 'interval' => 10000],
                // Medium polling for next 4 seconds (50ms)
                ['duration' => 4, 'interval' => 50000],
                // Slower polling for next 10 seconds (100ms)
                ['duration' => 10, 'interval' => 100000],
                // Very slow polling for the rest (250ms)
                ['duration' => $maxTimeout, 'interval' => 250000]
            ];
            
            $currentPollingStage = 0;
            $stageStartTime = $startTime;
            $currentInterval = $pollingIntervals[0]['interval'];
            $attempts = 0;
            SystemMessages::sysLogMsg(
                static::class,
                sprintf(
                    "Waiting for response on key: %s",
                    $responseKey,
                ),
                LOG_DEBUG
            );
            while (microtime(true) < $endTime) {
                try {
                    // Check if we need to adjust polling interval
                    $elapsedTime = microtime(true) - $startTime;

The assignment to $elapsedTime is dead and can be removed.

                    $stageElapsedTime = microtime(true) - $stageStartTime;
                    
                    if ($currentPollingStage < count($pollingIntervals) - 1 && 
                        $stageElapsedTime >= $pollingIntervals[$currentPollingStage]['duration']) {
                        // Move to next polling stage
                        $currentPollingStage++;
                        $stageStartTime = microtime(true);
                        $currentInterval = $pollingIntervals[$currentPollingStage]['interval'];
                        
                        // Skip logging interval changes to reduce noise
                    }
                    
                    // Check for response
                    $attempts++;
                    $encodedResponse = $redis->get($responseKey);
                    
                    if ($encodedResponse !== false) {
                        $responseTime = microtime(true);
                        $responseDelay = $responseTime - $startTime;
                        
                        // Response received - only log if it took longer than expected (over 1 second)
                        if ($responseDelay > 1.0) {
                            SystemMessages::sysLogMsg(
                                static::class,
                                sprintf(
                                    "Delayed response for action '%s' received after %.3fs (attempts: %d)",
                                    $actionName,
                                    $responseDelay,
                                    $attempts
                                ),
                                LOG_NOTICE
                            );
                        }
                        
                        $response = json_decode($encodedResponse, true);
                        break;
                    }
                    
                    // Short sleep before next check
                    usleep($currentInterval);
                    
                } catch (RedisException $redisException) {
                    // Specific handling for Redis exceptions
                    $retryCount++;
                    
                    if ($retryCount > $maxRetries) {
                        // Log last failed attempt before throwing
                        SystemMessages::sysLogMsg(
                            static::class,
                            sprintf(
                                "Max Redis retry attempts exceeded: %s (action: %s)",
                                $redisException->getMessage(),
                                $actionName
                            ),
                            LOG_ERR
                        );
                        throw $redisException; // Max retries exceeded
                    }
                    
                    // Log Redis-specific retry attempt
                    $errorMessage = sprintf(
                        "Redis connection error during worker request, retrying (%d/%d): %s (action: %s)",
                        $retryCount, 
                        $maxRetries, 
                        $redisException->getMessage(),
                        $actionName
                    );
                    
                    // Log to system log
                    SystemMessages::sysLogMsg(static::class, $errorMessage, LOG_WARNING);
                    
                    // Also log with error handler for potential Sentry capture
                    CriticalErrorsHandler::handleExceptionWithSyslog(
                        new \Exception($errorMessage)
                    );
                    
                    // Exponential backoff with jitter
                    $sleepTime = min(pow(2, $retryCount) * 100000, 2000000) + mt_rand(0, 100000);
                    usleep($sleepTime);
                    
                    // Reconnect
                    $redis = $this->di->get(RedisClientProvider::SERVICE_NAME);
                } catch (Throwable $otherException) {
                    // Handle other errors with retry
                    $retryCount++;
                    
                    if ($retryCount > $maxRetries) {
                        // Log last failed attempt before throwing
                        SystemMessages::sysLogMsg(
                            static::class,
                            sprintf(
                                "Max retry attempts exceeded for general exception: %s (action: %s)",
                                $otherException->getMessage(),
                                $actionName
                            ),
                            LOG_ERR
                        );
                        throw $otherException; // Max retries exceeded
                    }
                    
                    // Log general exception retry attempt
                    $errorMessage = sprintf(
                        "Error during worker request, retrying (%d/%d): %s (action: %s)",
                        $retryCount, 
                        $maxRetries, 
                        $otherException->getMessage(),
                        $actionName
                    );
                    
                    // Log to system log
                    SystemMessages::sysLogMsg(static::class, $errorMessage, LOG_WARNING);
                    
                    // Also log with error handler for potential Sentry capture
                    CriticalErrorsHandler::handleExceptionWithSyslog(
                        new \Exception($errorMessage)
                    );
                    
                    // Exponential backoff with jitter
                    $sleepTime = min(pow(2, $retryCount) * 100000, 2000000) + mt_rand(0, 100000);
                    usleep($sleepTime);
                    
                    // Reconnect
                    $redis = $this->di->get(RedisClientProvider::SERVICE_NAME);
                }
            }

            // Clean up
            try {
                $redis->del($responseKey);
            } catch (Throwable $e) {
                // Ignore cleanup errors
            }

            // Calculate total time taken
            $totalTime = microtime(true) - $startTime;

            if ($response === null) {
                SystemMessages::sysLogMsg(
                    static::class,
                    sprintf(
                        "Request timeout after %.3fs: No response received for action %s",
                        $totalTime,
                        $actionName
                    ),
                    LOG_WARNING
                );
                $this->response->setPayloadError('Request timeout or worker not responding');
                return;
            }

           // Handle file-based response
           if (isset($response[WorkerApiCommands::REDIS_RESPONSE_IN_FILE])) {
            $filename = $response[WorkerApiCommands::REDIS_RESPONSE_IN_FILE];
            if (file_exists($filename)) {
                $fileContent = file_get_contents($filename);
                if ($fileContent !== false) {
                    // Check if response is compressed
                    if (isset($response['compressed']) && $response['compressed'] === true) {
                        $fileContent = gzdecode($fileContent);
                    }
                    
                    $response = unserialize($fileContent);
                    unlink($filename);
                    $this->response->setPayloadSuccess($response);
                } else {
                    $this->response->setPayloadError('Failed to read response file');
                }
            } else {
                $this->response->setPayloadError('Response file not found');
            }
            return;
        }
        
        // Handle compressed Redis-based large response
        if (isset($response['large_response_redis'])) {
            $redisKey = $response['large_response_redis'];
            $compressedData = $redis->get($redisKey);
            
            if ($compressedData !== false) {
                // Clear the key since we're consuming the data
                $redis->del($redisKey);
                
                // Decompress and unserialize the data
                $data = gzdecode($compressedData);
                if ($data !== false) {
                    $response = unserialize($data);
                    $this->response->setPayloadSuccess($response);
                } else {
                    $this->response->setPayloadError('Failed to decompress response data');
                }
            } else {
                $this->response->setPayloadError('Large response data not found in Redis');
            }
            return;
        }


            if (array_key_exists(WorkerApiCommands::REDIS_JOB_PROCESSING_ERROR, $response)) {
                SystemMessages::sysLogMsg(
                    static::class,
                    sprintf(
                        "Setting error response: job_id=%s, error=%s",
                        $requestMessage['request_id'] ?? 'unknown',
                        $response[WorkerApiCommands::REDIS_JOB_PROCESSING_ERROR]
                    ),
                    LOG_DEBUG
                );
                $this->response->setPayloadError($response[WorkerApiCommands::REDIS_JOB_PROCESSING_ERROR]);
            } else {
                // Only log response time if it took longer than expected
                if ($totalTime > 1.0) {
                    SystemMessages::sysLogMsg(
                        static::class,
                        sprintf(
                            "Slow response for action '%s': %.3fs with %d attempts",
                            $actionName,
                            $totalTime,
                            $attempts
                        ),
                        LOG_NOTICE
                    );
                }
                $this->response->setPayloadSuccess($response);
            }

        } catch (Throwable $e) {
            // Log the error with detailed information
            $errorMessage = sprintf(
                "Error in sendRequestToBackendWorker: %s (processor: %s, action: %s)", 
                $e->getMessage(),
                $processor,
                $actionName
            );
            
            // Log to system log first for visibility in server logs
            SystemMessages::sysLogMsg(static::class, $errorMessage, LOG_ERR);
            
            // Then use critical error handler for potential Sentry capture
            CriticalErrorsHandler::handleExceptionWithSyslog($e);
            
            // Return error to client
            $this->response->setPayloadError($e->getMessage());
        }