Code Coverage
 
Lines
Functions and Methods
Classes and Traits
Total
97.78% covered (success)
97.78%
44 / 45
83.33% covered (warning)
83.33%
5 / 6
 CRAP
0.00% covered (danger)
0.00%
0 / 1
LocalWorkerCoordinator
97.78% covered (success)
97.78%
44 / 45
83.33% covered (warning)
83.33%
5 / 6
 20
0.00% covered (danger)
0.00%
0 / 1
 __construct
100.00% covered (success)
100.00%
1 / 1
100.00% covered (success)
100.00%
1 / 1
 1
 ensureCapacity
100.00% covered (success)
100.00%
6 / 6
100.00% covered (success)
100.00%
1 / 1
 3
 launchUnderLock
100.00% covered (success)
100.00%
13 / 13
100.00% covered (success)
100.00%
1 / 1
 5
 runningWorkers
88.89% covered (warning)
88.89%
8 / 9
0.00% covered (danger)
0.00%
0 / 1
 4.02
 workersToStart
100.00% covered (success)
100.00%
3 / 3
100.00% covered (success)
100.00%
1 / 1
 3
 startWorkers
100.00% covered (success)
100.00%
13 / 13
100.00% covered (success)
100.00%
1 / 1
 4
1<?php
2
3declare(strict_types=1);
4/**
5 * SPDX-FileCopyrightText: 2026 LibreCode coop and contributors
6 * SPDX-License-Identifier: AGPL-3.0-or-later
7 */
8
9namespace OCA\Libresign\Service\Worker;
10
11use Psr\Log\LoggerInterface;
12
13/** Best-effort local launch coordination; not process supervision. */
14class LocalWorkerCoordinator {
15    private const int COOLDOWN_SECONDS = 2;
16
17    public function __construct(
18        private WorkerStarter $starter,
19        private WorkerJobCounter $jobCounter,
20        private LocalWorkerProcess $processes,
21        private LocalWorkerStateStore $store,
22        private LoggerInterface $logger,
23    ) {
24    }
25
26    public function ensureCapacity(int $desired, string $pool = WorkerConfiguration::SIGNING): bool {
27        if ($desired <= 0) {
28            return true;
29        }
30        if (!$this->processes->isSupported()) {
31            $this->logger->warning('Local workers require readable Linux procfs; use Nextcloud cron or dedicated workers');
32            return false;
33        }
34
35        return $this->store->withExclusiveLock(fn (): bool => $this->launchUnderLock(min($desired, 32), $pool), $pool);
36    }
37
38    private function launchUnderLock(int $desired, string $pool): bool {
39        $state = $this->store->readState($pool);
40        $running = $this->runningWorkers($state['workers'] ?? [], $this->jobCounter->jobClasses($pool));
41        $availableSlots = max(0, $desired - count($running));
42        $toStart = $this->workersToStart($availableSlots, $state, $pool);
43
44        if ($toStart > 0) {
45            $before = count($running);
46            $running = $this->startWorkers($toStart, $running, $pool);
47            // Keep signing's existing cooldown. For notifications, only a
48            // failed/unidentified start needs one; drained workers may restart.
49            if ($pool === WorkerConfiguration::SIGNING || count($running) === $before) {
50                $state['lastAttempt'] = microtime(true);
51            } else {
52                unset($state['lastAttempt']);
53            }
54        }
55
56        $state['workers'] = $running;
57        $this->store->writeState($state, $pool);
58        return $running !== [] || $toStart === 0;
59    }
60
61    /** @param array<mixed> $entries
62     * @param list<class-string> $jobClasses
63     * @return list<array{pid: int, start: string}>
64     */
65    private function runningWorkers(array $entries, array $jobClasses): array {
66        $running = [];
67        foreach ($entries as $entry) {
68            if (!is_array($entry)) {
69                continue;
70            }
71            $pid = (int)($entry['pid'] ?? 0);
72            $startTime = (string)($entry['start'] ?? '');
73            if ($this->processes->isRunning($pid, $startTime, $jobClasses)) {
74                $running[] = ['pid' => $pid, 'start' => $startTime];
75            }
76        }
77        return $running;
78    }
79
80    /** @param array<string, mixed> $state */
81    private function workersToStart(int $availableSlots, array $state, string $pool): int {
82        if ($availableSlots === 0 || microtime(true) - (float)($state['lastAttempt'] ?? 0) < self::COOLDOWN_SECONDS) {
83            return 0;
84        }
85        return min($availableSlots, $this->jobCounter->countEligibleJobs($availableSlots, $pool));
86    }
87
88    /** @param list<array{pid: int, start: string}> $running
89     * @return list<array{pid: int, start: string}>
90     */
91    private function startWorkers(int $count, array $running, string $pool): array {
92        for ($i = 0; $i < $count; $i++) {
93            try {
94                $this->logger->debug('Local worker launch attempt', ['pool' => $pool]);
95                $pid = $this->starter->startWorker($pool);
96                $startTime = $this->processes->startTime($pid);
97                if ($startTime === null) {
98                    $this->logger->warning('Worker process identity is unavailable', ['pid' => $pid]);
99                    // The child may still be running. Do not spawn additional
100                    // workers without a verified identity.
101                    break;
102                }
103                $running[] = ['pid' => $pid, 'start' => $startTime];
104                $this->logger->debug('Local worker process identified', ['pool' => $pool, 'pid' => $pid]);
105            } catch (\RuntimeException $e) {
106                $this->logger->warning('Cannot start local ' . $pool . ' worker', ['exception' => $e]);
107                break;
108            }
109        }
110        return $running;
111    }
112}