Lines 100.00% 42 / 42
Methods 100.00% 7 / 7
Classes 100.00% 1 / 1
Name Lines Methods CRAP
 __construct 100.00% 1 / 1 100.00% 1 / 1 1
 startProbing 100.00% 10 / 10 100.00% 1 / 1 3
 stopProbing 100.00% 2 / 2 100.00% 1 / 1 1
 probeWorkers 100.00% 14 / 14 100.00% 1 / 1 5
 recordPong 100.00% 7 / 7 100.00% 1 / 1 2
 countMissedProbes 100.00% 1 / 1 100.00% 1 / 1 1
 killUnresponsive 100.00% 7 / 7 100.00% 1 / 1 1
17final class WorkerWatchdog
18{
19    private const int REMIND_EVERY_NTH_MISSED_PROBE = 6;
20
21    private ?RepeatingTimer $timer = null;
22
23    private readonly EveryNthOccurrence $unresponsiveReminders;
24
25    public function __construct(
26        private readonly WorkerSlotRegistry $slots,
27        private readonly WorkerProbeSequence $probes,
28        private readonly SupervisionConfig $supervision,
29        private readonly Timers $timers,
30        private readonly Clock $clock,
31        private readonly LoggerInterface $logger,
32    ) {
33        $this->unresponsiveReminders = new EveryNthOccurrence(self::REMIND_EVERY_NTH_MISSED_PROBE);
34    }
35
36    public function startProbing(): void
37    {
38        $interval = $this->supervision->pingInterval->findDuration();
39
40        if ($interval === null || $this->timer !== null) {
41            return;
42        }
43
44        $this->timer = new RepeatingTimer(
45            $this->timers,
46            $interval,
47            $this->probeWorkers(...),
48            fn(Throwable $e) => $this->logger->warning('Watchdog probe failed', ['exception' => $e]),
49        );
50        $this->timer->start();
51    }
52
53    public function stopProbing(): void
54    {
55        $this->timer?->stop();
56        $this->timer = null;
57    }
58
59    public function probeWorkers(): void
60    {
61        $nonce = $this->probes->advanceNonce();
62
63        foreach ($this->slots->listLiveHandles() as $handle) {
64            $missed = $handle->countMissedProbes($nonce);
65
66            if ($this->supervision->killsUnresponsiveWorkers() && $missed >= $this->supervision->unresponsiveAfter) {
67                $this->killUnresponsive($handle, $missed);
68
69                continue;
70            }
71
72            if ($this->unresponsiveReminders->includesOccurrence($missed)) {
73                $this->logger->warning('Worker is not answering liveness probes', [
74                    'worker' => $handle->id->value,
75                    'pid' => $handle->getPid(),
76                    'apps' => $handle->getAppIds()->toStrings(),
77                    'missed_probes' => $missed,
78                ]);
79            }
80
81            $handle->send(new Ping($nonce, $this->clock->getNow()));
82        }
83    }
84
85    public function recordPong(WorkerHandle $handle, Pong $pong): void
86    {
87        $handle->recordPong($pong, $this->clock->getNow());
88
89        if ($pong->loopLag->isLongerThan($this->supervision->lagThreshold)) {
90            $this->logger->warning('Worker event loop is lagging; is an app blocking?', [
91                'worker' => $handle->id->value,
92                'lag' => (string) $pong->loopLag,
93                'apps' => $handle->getAppIds()->toStrings(),
94            ]);
95        }
96    }
97
98    public function countMissedProbes(WorkerHandle $handle): int
99    {
100        return $handle->countMissedProbes($this->probes->getCurrentNonce() + 1);
101    }
102
103    private function killUnresponsive(WorkerHandle $handle, int $missed): void
104    {
105        $this->logger->error('Worker stopped answering liveness probes; killing it', [
106            'worker' => $handle->id->value,
107            'pid' => $handle->getPid(),
108            'apps' => $handle->getAppIds()->toStrings(),
109            'missed_probes' => $missed,
110        ]);
111
112        $handle->terminate();
113    }
114}