refactor: improve worker process management and logging

- Add support for forked process PID file management
- Implement proper PID file handling for child processes
- Add setForked() method to track forked state
- Enhance debug logging in WorkerSafeScriptsCore
- Improve error handling and response management in WorkerApiCommands
- Add child process timeout handling
- Refactor request processing into smaller, more maintainable methods
- Follow PSR-12 code style standards
- Add comprehensive English comments

Author: jorikfon

src/Core/Workers/Cron/WorkerSafeScriptsCore.php

@@ -263,6 +263,7 @@ private function processStartBatch(array $batch, array &$childPids): void
             if ($pid === 0) {
                 // Child process
                 try {
+                    $this->setForked();
                     $this->checkWorkerByType($workerConfig['type'], $workerConfig['worker']);
                     exit(0);
                 } catch (Throwable $e) {
@@ -291,6 +292,7 @@ private function processBatch(array $batch, array &$childPids): void
             if ($pid === 0) {
                 // Child process
                 try {
+                    $this->setForked();
                     SystemMessages::sysLogMsg(
                         __CLASS__,
                         "Restarting worker: $worker",
@@ -464,8 +466,11 @@ public function checkWorkerBeanstalk(string $workerClassName): void
             $result = false;
             if ($WorkerPID !== '') {
                 // Ping the worker via Beanstalk queue.
+                SystemMessages::sysLogMsg(__METHOD__, "Service $workerClassName is alive. Sending ping request.", LOG_DEBUG);
                 $queue = new BeanstalkClient($this->makePingTubeName($workerClassName));
                 [$result] = $queue->sendRequest('ping', 5, 1);
+                SystemMessages::sysLogMsg(__METHOD__, "Service $workerClassName answered $result", LOG_DEBUG);
+
             }
             if (false === $result) {
                 Processes::processPHPWorker($workerClassName);

src/Core/Workers/WorkerBase.php

@@ -89,6 +89,13 @@ abstract class WorkerBase extends Injectable implements WorkerInterface
 
     private const string LOG_NAMESPACE_SEPARATOR = '\\';
 
+    /**
+     * Flag indicating whether the worker is a forked process
+     *
+     * @var bool
+     */
+    private bool $isForked = false;
+
     /**
      * Maximum number of processes that can be created
      *
@@ -203,17 +210,52 @@ protected function setWorkerState(int $state): void
      *
      * @throws RuntimeException If unable to write PID file
      */
+    /**
+     * Handles PID file operations for worker processes
+     */
     private function savePidFile(): void
     {
         try {
-            $activeProcesses = Processes::getPidOfProcess(static::class);
-            $processes = array_filter(explode(' ', $activeProcesses));
+            $pid = getmypid();
+            if ($pid === false) {
+                throw new RuntimeException('Could not get process ID');
+            }
+
+            $pidFile = $this->getPidFile();
+            $pidDir = dirname($pidFile);
 
-            if (count($processes) === 1) {
-                $this->saveSinglePidFile($activeProcesses);
-            } else {
-                $this->saveMultiplePidFiles($processes);
+            // Ensure PID directory exists
+            if (!is_dir($pidDir) && !mkdir($pidDir, 0755, true)) {
+                throw new RuntimeException("Could not create PID directory: $pidDir");
             }
+
+            // For forked processes, append the PID to the filename
+            if (isset($this->isForked) && $this->isForked === true) {
+                $pidFile = $this->getForkedPidFile($pid);
+            }
+
+            // Use exclusive file creation to avoid race conditions
+            $handle = fopen($pidFile, 'c+');
+            if ($handle === false) {
+                throw new RuntimeException("Could not open PID file: $pidFile");
+            }
+
+            if (!flock($handle, LOCK_EX | LOCK_NB)) {
+                fclose($handle);
+                throw new RuntimeException("Could not acquire lock on PID file: $pidFile");
+            }
+
+            // Write PID to file
+            if (ftruncate($handle, 0) === false ||
+                fwrite($handle, (string)$pid) === false) {
+                flock($handle, LOCK_UN);
+                fclose($handle);
+                throw new RuntimeException("Could not write to PID file: $pidFile");
+            }
+
+            flock($handle, LOCK_UN);
+            fclose($handle);
+
         } catch (Throwable $e) {
             SystemMessages::sysLogMsg(
                 static::class,
@@ -225,51 +267,30 @@ private function savePidFile(): void
     }
 
     /**
-     * Saves a single PID to file
-     *
-     * @param string $pid Process ID to save
-     * @throws RuntimeException If write fails
-     */
-    private function saveSinglePidFile(string $pid): void
-    {
-        if (!file_put_contents($this->getPidFile(), $pid)) {
-            throw new RuntimeException('Could not write to PID file');
-        }
-    }
-
-    /**
-     * Saves multiple PIDs to separate files
+     * Generate the PID file path for the worker
      *
-     * @param array $processes Array of process IDs
-     * @throws RuntimeException If write fails
+     * @return string The path to the PID file
      */
-    private function saveMultiplePidFiles(array $processes): void
+    public function getPidFile(): string
     {
-        $pidFilesDir = dirname($this->getPidFile());
-        $baseName = (string)pathinfo($this->getPidFile(), PATHINFO_BASENAME);
-        $pidFile = $pidFilesDir . '/' . $baseName;
-
-        // Delete old PID files
-        $rm = Util::which('rm');
-        Processes::mwExec("$rm -rf $pidFile*");
-
-        foreach ($processes as $index => $process) {
-            $pidFilePath = sprintf("%s-%d%s", $pidFile, $index + 1, self::PID_FILE_SUFFIX);
-            if (!file_put_contents($pidFilePath, $process)) {
-                throw new RuntimeException("Could not write to PID file: $pidFilePath");
-            }
-        }
+        $name = str_replace("\\", '-', static::class);
+        return self::PID_FILE_DIR . "/$name" . self::PID_FILE_SUFFIX;
     }
 
     /**
-     * Generate the PID file path for the worker
+     * Generates PID file path for forked processes
      *
-     * @return string The path to the PID file
+     * @param int $pid Process ID
+     * @return string Full path to the PID file
      */
-    public function getPidFile(): string
+    private function getForkedPidFile(int $pid): string
     {
-        $name = str_replace("\\", '-', static::class);
-        return self::PID_FILE_DIR . "/$name" . self::PID_FILE_SUFFIX;
+        $basePidFile = $this->getPidFile();
+        return sprintf(
+            '%s.%d',
+            $basePidFile,
+            $pid
+        );
     }
 
     /**
@@ -425,14 +446,27 @@ private function logErrorShutdown(string $processTitle, float $timeElapsedSecs,
     }
 
     /**
-     * Cleans up PID file during shutdown
+     * Safely cleans up PID file during shutdown
      */
     private function cleanupPidFile(): void
     {
         try {
-            $pidFile = $this->getPidFile();
+            $pid = getmypid();
+            if ($pid === false) {
+                return;
+            }
+
+            // Determine which PID file to remove
+            $pidFile = isset($this->isForked) && $this->isForked === true
+                ? $this->getForkedPidFile($pid)
+                : $this->getPidFile();
+
+            // Only remove the file if it exists and contains our PID
             if (file_exists($pidFile)) {
-                unlink($pidFile);
+                $storedPid = file_get_contents($pidFile);
+                if ($storedPid === (string)$pid) {
+                    unlink($pidFile);
+                }
             }
         } catch (Throwable $e) {
             SystemMessages::sysLogMsg(
@@ -541,6 +575,14 @@ public function makePingTubeName(string $workerClassName): string
         return Text::camelize("ping_$workerClassName", '\\');
     }
 
+    /**
+     * Sets flag for forked process after pcntl_fork()
+     */
+    public function setForked(): void
+    {
+        $this->isForked = true;
+    }
+
     /**
      * Destructor - ensures PID file is saved on object destruction
      */

src/Core/Workers/WorkerPrepareAdvice.php

@@ -130,6 +130,7 @@ public function start(array $argv): void
             } elseif ($pid == 0) {
                 // Child process.
                 try {
+                    $this->setForked();
                     $this->processAdvice($adviceType);
                 } catch (Throwable $e) {
                     CriticalErrorsHandler::handleExceptionWithSyslog($e);

src/PBXCoreREST/Workers/WorkerApiCommands.php

@@ -20,6 +20,8 @@
 
 namespace MikoPBX\PBXCoreREST\Workers;
 
+use InvalidArgumentException;
+use JsonException;
 use MikoPBX\Common\Handlers\CriticalErrorsHandler;
 use MikoPBX\Common\Providers\BeanstalkConnectionWorkerApiProvider;
 use MikoPBX\Core\System\{BeanstalkClient, Configs\BeanstalkConf, Directories, Processes, SystemMessages};
@@ -28,8 +30,8 @@
 use MikoPBX\PBXCoreREST\Lib\PBXApiResult;
 use MikoPBX\PBXCoreREST\Lib\PbxExtensionsProcessor;
 use MikoPBX\PBXCoreREST\Lib\SystemManagementProcessor;
+use RuntimeException;
 use Throwable;
-
 use function xdebug_break;
 
 require_once 'Globals.php';
@@ -47,11 +49,9 @@
 class WorkerApiCommands extends WorkerBase
 {
     /**
-     * The maximum parallel worker processes
-     *
-     * @var int
+     * Maximum time to wait for child process (seconds)
      */
-    public int $maxProc = 4;
+    private const int CHILD_PROCESS_TIMEOUT = 60;
 
 
     /**
@@ -92,77 +92,265 @@ public function prepareAnswer(BeanstalkClient $message): void
             // Fork failed
             throw new \RuntimeException("Failed to fork a new process.");
         }
+
         if ($pid === 0) {
-            $res = new PBXApiResult();
-            $res->processor = __METHOD__;
-            $async = false;
             try {
-                $request = json_decode($message->getBody(), true, 512, JSON_THROW_ON_ERROR);
-                $async   = ($request['async'] ?? false) === true;
-                $processor = $request['processor'];
-                $res->processor = $processor;
-                // Old style, we can remove it in 2025
-                if ($processor === 'modules') {
-                    $processor = PbxExtensionsProcessor::class;
-                }
+                // Child process
+                $this->setForked();
+                $this->processRequest($message);
+            } catch (Throwable $e) {
+                CriticalErrorsHandler::handleExceptionWithSyslog($e);
+                exit(1); // Exit with error
+            }
+            exit(0);
+        }
 
-                $this->breakpointHere($request);
-
-                // This is the child process
-                if (method_exists($processor, 'callback')) {
-                    cli_set_process_title(__CLASS__ . '-' . $request['action']);
-                    // Execute async job
-                    if ($async) {
-                        $res->success = true;
-                        $res->messages['info'][] = "The async job {$request['action']} starts in background, you will receive answer on {$request['asyncChannelId']} nchan channel";
-                        $encodedResult = json_encode($res->getResult());
-                        $message->reply($encodedResult);
-                        $processor::callback($request);
-                    } else {
-                        $res = $processor::callback($request);
-                    }
-                } else {
-                    $res->success = false;
-                    $res->messages['error'][] = "Unknown processor - $processor in prepareAnswer";
-                }
-            } catch (Throwable $exception) {
-                $request = [];
-                $this->needRestart = true;
-                // Prepare answer with pretty error description
-                $res->messages['error'][] = CriticalErrorsHandler::handleExceptionWithSyslog($exception);
-            } finally {
-                if ($async === false) {
-                    $encodedResult = json_encode($res->getResult());
-                    if ($encodedResult === false) {
-                        $res->data = [];
-                        $res->messages['error'][] = 'It is impossible to encode to json current processor answer';
-                        $encodedResult = json_encode($res->getResult());
-                    }
+        // Parent process
+        $startTime = time();
+        $status = 0;
 
-                    // Check the response size and write in on file if it bigger than Beanstalk can digest
-                    if (strlen($encodedResult) > BeanstalkConf::JOB_DATA_SIZE_LIMIT) {
-                        $downloadCacheDir = Directories::getDir(Directories::WWW_DOWNLOAD_CACHE_DIR);
-                        $filenameTmp = $downloadCacheDir . '/temp-' . __FUNCTION__ . '_' . microtime() . '.data';
-                        if (file_put_contents($filenameTmp, serialize($res->getResult()))) {
-                            $encodedResult = json_encode([BeanstalkClient::RESPONSE_IN_FILE => $filenameTmp]);
-                        } else {
-                            $res->data = [];
-                            $res->messages['error'][] = 'It is impossible to write answer into file ' . $filenameTmp;
-                            $encodedResult = json_encode($res->getResult());
-                        }
+        // Wait for child with timeout
+        while (time() - $startTime < self::CHILD_PROCESS_TIMEOUT) {
+            $res = pcntl_waitpid($pid, $status, WNOHANG);
+            if ($res === -1) {
+                throw new RuntimeException("Failed to wait for child process");
+            }
+            if ($res > 0) {
+                // Child process completed
+                if (pcntl_wifexited($status)) {
+                    $exitStatus = pcntl_wexitstatus($status);
+                    if ($exitStatus !== 0) {
+                        throw new RuntimeException("Child process failed with status: $exitStatus");
                     }
-                    $message->reply($encodedResult);
+                    return;
                 }
-                if ($res->success) {
-                    $this->checkNeedReload($request);
+                if (pcntl_wifsignaled($status)) {
+                    $signal = pcntl_wtermsig($status);
+                    throw new RuntimeException("Child process terminated by signal: $signal");
                 }
+                return;
+            }
+            usleep(100000); // Sleep 100ms
+        }
+
+        // Timeout reached
+        posix_kill($pid, SIGTERM);
+        throw new RuntimeException("Child process timed out");
+    }
+
+    /**
+     * Process individual API request
+     *
+     * @param BeanstalkClient $message The message from beanstalk queue
+     *
+     * @throws JsonException If JSON parsing fails
+     * @throws RuntimeException|Throwable If processor execution fails
+     */
+    private function processRequest(BeanstalkClient $message): void
+    {
+        $res = new PBXApiResult();
+        $res->processor = __METHOD__;
+        try {
+            // Parse request JSON
+            $request = $this->parseRequestJson($message);
+
+            // Setup basic request parameters
+            $async = (bool)($request['async'] ?? false);
+            $processor = $this->resolveProcessor($request);
+
+            $res->processor = $processor;
+
+            // Old style, we can remove it in 2025
+            if ($processor === 'modules') {
+                $processor = PbxExtensionsProcessor::class;
+            }
+
+            // Handle debug mode if needed
+            $this->handleDebugMode($request);
+
+            // Process the request
+            if (!method_exists($processor, 'callback')) {
+                throw new RuntimeException("Unknown processor - {$processor}");
+            }
+
+            cli_set_process_title(__CLASS__ . '-' . $request['action']);
+
+            // Execute request based on async flag
+            if ($async) {
+                $this->handleAsyncRequest($message, $request, $res);
+            } else {
+                $res = $processor::callback($request);
+                $this->sendResponse($message, $res);
+            }
+
+            // Check if reload is needed after successful execution
+            if ($res->success) {
+                $this->checkNeedReload($request);
+            }
+
+        } catch (JsonException $e) {
+            $this->handleError($res, "Invalid JSON in request: {$e->getMessage()}");
+            $this->sendResponse($message, $res);
+        } catch (InvalidArgumentException $e) {
+            $this->handleError($res, "Invalid request parameters: {$e->getMessage()}");
+            $this->sendResponse($message, $res);
+        } catch (Throwable $e) {
+            $this->handleError($res, CriticalErrorsHandler::handleExceptionWithSyslog($e));
+            $this->sendResponse($message, $res);
+            throw $e; // Re-throw for parent process to handle
+        }
+    }
+
+    /**
+     * Parse and validate request JSON
+     *
+     * @param BeanstalkClient $message
+     * @return array
+     * @throws JsonException
+     */
+    private function parseRequestJson(BeanstalkClient $message): array
+    {
+        $request = json_decode(
+            $message->getBody(),
+            true,
+            512,
+            JSON_THROW_ON_ERROR
+        );
+
+        if (!is_array($request)) {
+            throw new InvalidArgumentException('Request must be a JSON object');
+        }
+
+        return $request;
+    }
+
+    /**
+     * Resolve processor class name
+     *
+     * @param array $request
+     * @return string
+     * @throws InvalidArgumentException
+     */
+    private function resolveProcessor(array $request): string
+    {
+        $processor = $request['processor'] ?? '';
+
+        // Handle legacy 'modules' processor name
+        if ($processor === 'modules') {
+            return PbxExtensionsProcessor::class;
+        }
+
+        if (empty($processor)) {
+            throw new InvalidArgumentException('Processor name is required');
+        }
+
+        return $processor;
+    }
+
+    /**
+     * Handle asynchronous request execution
+     *
+     * @param BeanstalkClient $message
+     * @param array $request
+     * @param PBXApiResult $res
+     */
+    private function handleAsyncRequest(
+        BeanstalkClient $message,
+        array           $request,
+        PBXApiResult    $res
+    ): void
+    {
+        $res->success = true;
+        $res->messages['info'][] = sprintf(
+            'The async job %s starts in background, you will receive answer on %s nchan channel',
+            $request['action'],
+            $request['asyncChannelId']
+        );
+
+        $this->sendResponse($message, $res);
+        $request['processor']::callback($request);
+    }
+
+    /**
+     * Send response back through beanstalk
+     *
+     * @param BeanstalkClient $message
+     * @param PBXApiResult $res
+     * @throws RuntimeException
+     */
+    private function sendResponse(BeanstalkClient $message, PBXApiResult $res): void
+    {
+        try {
+            $result = $res->getResult();
+            $encodedResult = json_encode($result);
+
+            if ($encodedResult === false) {
+                $res->data = [];
+                $res->messages['error'][] = 'Failed to encode response to JSON';
+                $encodedResult = json_encode($res->getResult());
+            }
+
+            // Handle large responses
+            if (strlen($encodedResult) > BeanstalkConf::JOB_DATA_SIZE_LIMIT) {
+                $encodedResult = $this->handleLargeResponse($result);
             }
-            exit(0); // Exit the child process
+
+            $message->reply($encodedResult);
+
+        } catch (Throwable $e) {
+            throw new RuntimeException(
+                "Failed to send response: {$e->getMessage()}",
+                0,
+                $e
+            );
+        }
+    }
+
+    /**
+     * Handle large response by storing in temporary file
+     *
+     * @param array $result
+     * @return string JSON encoded response with file reference
+     * @throws RuntimeException
+     */
+    private function handleLargeResponse(array $result): string
+    {
+        $downloadCacheDir = Directories::getDir(Directories::WWW_DOWNLOAD_CACHE_DIR);
+
+        // Generate unique filename using uniqid()
+        $filenameTmp = sprintf(
+            '%s/temp-%s_%s.data',
+            $downloadCacheDir,
+            __FUNCTION__,
+            uniqid('', true)
+        );
+
+        // Check available disk space
+        if (disk_free_space($downloadCacheDir) < strlen(serialize($result))) {
+            throw new RuntimeException('Insufficient disk space for temporary file');
+        }
+
+        if (!file_put_contents($filenameTmp, serialize($result))) {
+            throw new RuntimeException("Failed to write response to temporary file");
         }
-        // This is the parent process
-        pcntl_wait($status); // Wait for the child process to complete
+
+        return json_encode([BeanstalkClient::RESPONSE_IN_FILE => $filenameTmp]);
+    }
+
+    /**
+     * Handle error cases
+     *
+     * @param PBXApiResult $res
+     * @param string $message
+     */
+    private function handleError(PBXApiResult $res, string $message): void
+    {
+        $res->success = false;
+        $res->messages['error'][] = $message;
+        $res->data = [];
     }
 
+
     /**
      * Checks if the module or worker needs to be reloaded.
      *
@@ -230,7 +418,7 @@ private function getNeedRestartActions(): array
      *      ...
      * });
      */
-    private function breakpointHere(array $request): void
+    private function handleDebugMode(array $request): void
     {
         if (isset($request['debug']) && $request['debug'] === true && extension_loaded('xdebug')) {
             if (function_exists('xdebug_connect_to_client')) {