Lines 97.03% 131 / 135
Functions and Methods 81.25% 13 / 16
Classes and Traits 0.00% 0 / 1
Name Lines Functions and Methods CRAP Classes and Traits
WorkerPool 97.03% 131 / 135 81.25% 13 / 16 47 0.00% 0 / 1
 __construct 100.00% 1 / 1 100.00% 1 / 1 1
 startWorker 100.00% 12 / 12 100.00% 1 / 1 4
 restartWorker 100.00% 10 / 10 100.00% 1 / 1 6
 markReady 100.00% 6 / 6 100.00% 1 / 1 3
 shutdown 100.00% 12 / 12 100.00% 1 / 1 5
 terminateWorkers 100.00% 2 / 2 100.00% 1 / 1 2
 spawnProcessOrScheduleRestart 100.00% 11 / 11 100.00% 1 / 1 3
 spawnProcess 100.00% 4 / 4 100.00% 1 / 1 2
 attachHandle 100.00% 10 / 10 100.00% 1 / 1 1
 awaitWorkerExit 100.00% 9 / 9 100.00% 1 / 1 3
 armReadyDeadline 92.85% 13 / 14 0.00% 0 / 1 3.00
 onWorkerGone 91.66% 11 / 12 0.00% 0 / 1 5.01
 logUnexpectedExit 100.00% 6 / 6 100.00% 1 / 1 3
 scheduleRestart 100.00% 18 / 18 100.00% 1 / 1 2
 respawnOnRequest 100.00% 5 / 5 100.00% 1 / 1 2
 respawnWorker 33.33% 1 / 3 0.00% 0 / 1 3.19
1<?php
2
3declare(strict_types=1);
4
5namespace Stewart\Runtime\Broker;
6
7use Amp\Cancellation;
8use Amp\CancelledException;
9use Psr\Log\LoggerInterface;
10use Stewart\Contracts\Time\Clock;
11use Stewart\Contracts\Time\Duration;
12use Stewart\Contracts\Time\TimerHandle;
13use Stewart\Contracts\Time\Timers;
14use Stewart\Runtime\Config\StewartConfig;
15use Stewart\Runtime\Config\SupervisionConfig;
16use Stewart\Runtime\Exception\BrokerError;
17use Stewart\Runtime\Exception\BrokerException;
18use Stewart\Runtime\Exception\WorkerError;
19use Stewart\Runtime\Exception\WorkerException;
20use Stewart\Runtime\Ipc\Message\Shutdown;
21use Stewart\Runtime\Ipc\Message\WorkerMessage;
22use Stewart\Runtime\Lifecycle\WorkerPhase;
23use Stewart\Runtime\Model\WorkerId;
24use Stewart\Support\Time\Deadlines;
25use Throwable;
26
27use function Amp\async;
28use function Amp\Future\awaitAll;
29
30final class WorkerPool
31{
32    private const int JOIN_MARGIN_MILLISECONDS = 1_000;
33
34    private const int SPAWN_TIMEOUT_SECONDS = 10;
35
36    private const int EXIT_RESULT_WAIT_SECONDS = 2;
37
38    private bool $running = true;
39
40    public function __construct(
41        private readonly WorkerSpawner $spawner,
42        private readonly WorkerSlotRegistry $slots,
43        private readonly WorkerRestartPolicy $restartPolicy,
44        private readonly WorkerProbeSequence $probes,
45        private readonly SupervisionConfig $supervision,
46        private readonly WorkerMessageReader $messageReader,
47        private readonly LoggerInterface $logger,
48        private readonly OutboxLimits $outboxLimits,
49        private readonly Timers&Deadlines $timers,
50        private readonly Clock $clock,
51        private readonly StewartConfig $config,
52    ) {}
53
54    public function startWorker(WorkerSlot $slot, WorkerPoolListener $listener): bool
55    {
56        if (!$this->running) {
57            return false;
58        }
59
60        $state = $this->slots->findOrCreateSlotState($slot, $listener);
61        $state->beginSpawn();
62        $process = $this->spawnProcessOrScheduleRestart($state, $slot);
63
64        if ($process === null) {
65            return false;
66        }
67
68        if (!$state->isSpawning()) {
69            $process->close();
70
71            return false;
72        }
73
74        $this->attachHandle($state, $slot, $process, $listener);
75
76        return true;
77    }
78
79    /** @throws WorkerException */
80    public function restartWorker(WorkerSlot $slot): WorkerRestartOutcome
81    {
82        if (!$this->running) {
83            throw WorkerException::createForReason(WorkerError::ShuttingDown, workerId: $slot->workerId->value);
84        }
85
86        $state = $this->slots->findSlotState($slot->workerId);
87
88        // No state yet means startup has not reached this worker.
89        if ($state === null || $state->isSpawning() || $state->isStopping()) {
90            return WorkerRestartOutcome::AlreadyUnderway;
91        }
92
93        $this->logger->info('Restarting worker on request', ['worker' => $slot->workerId->value, 'apps' => $slot->listAppIds()->toStrings(), 'phase' => $state->phase->value]);
94        $this->restartPolicy->forgetRestarts($slot->workerId);
95        $stopping = $state->phase === WorkerPhase::Live ? $state->beginStopping() : null;
96
97        async(fn() => $this->respawnOnRequest($state, $stopping))->ignore();
98
99        return WorkerRestartOutcome::Begun;
100    }
101
102    public function markReady(WorkerHandle $handle): void
103    {
104        if (!$handle->isStarting()) {
105            return;
106        }
107
108        $handle->markReady();
109
110        $state = $this->slots->findSlotState($handle->id);
111
112        if ($state?->isLive($handle) === true) {
113            $state->cancelReadyDeadline();
114        }
115    }
116
117    public function shutdown(string $reason, Duration $grace): void
118    {
119        $this->running = false;
120        $states = $this->slots->listSlotStates();
121
122        foreach ($states as $state) {
123            $state->stop();
124        }
125
126        $deadline = $this->timers->timeout($grace->plus(Duration::milliseconds(self::JOIN_MARGIN_MILLISECONDS)));
127        $joins = [];
128
129        foreach ($states as $state) {
130            $handle = $state->handle;
131
132            if ($handle === null || $handle->isTerminated()) {
133                continue;
134            }
135
136            $handle->sendShutdown(new Shutdown($reason, $grace));
137            $joins[] = async(fn() => $this->awaitWorkerExit($handle, $deadline));
138        }
139
140        awaitAll($joins);
141    }
142
143    public function terminateWorkers(): void
144    {
145        foreach ($this->slots->listSlotStates() as $state) {
146            $state->handle?->terminate();
147        }
148    }
149
150    private function spawnProcessOrScheduleRestart(WorkerSlotState $state, WorkerSlot $slot): ?WorkerProcess
151    {
152        try {
153            return $this->spawnProcess($slot->workerId);
154        } catch (Throwable $e) {
155            $this->logger->error('Could not start worker; its automations are not running', [
156                'worker' => $slot->workerId->value,
157                'apps' => $slot->listAppIds()->toStrings(),
158                'exception' => $e,
159            ]);
160
161            // shutdown() may stop the slot while spawn() is suspended.
162            if ($state->isSpawning()) {
163                $state->markSpawnFailed();
164                $this->scheduleRestart($state);
165            }
166
167            return null;
168        }
169    }
170
171    /** @throws Throwable */
172    private function spawnProcess(WorkerId $workerId): WorkerProcess
173    {
174        $timeout = Duration::seconds(self::SPAWN_TIMEOUT_SECONDS);
175
176        try {
177            return $this->spawner->spawn($workerId, $this->timers->timeout($timeout));
178        } catch (CancelledException $e) {
179            throw BrokerException::createForReason(BrokerError::WorkerSpawnTimedOut, $e, worker: $workerId->value, timeout: (string) $timeout);
180        }
181    }
182
183    private function attachHandle(WorkerSlotState $state, WorkerSlot $slot, WorkerProcess $process, WorkerPoolListener $listener): void
184    {
185        $handle = new WorkerHandle($slot->workerId, $process, $slot, $this->logger, $this->outboxLimits, $this->probes->getCurrentNonce());
186        $state->recordSpawnedHandle($handle, $this->armReadyDeadline($handle))?->terminate();
187
188        $handle->relayOutput();
189        $this->messageReader->readMessagesInBackground(
190            $handle,
191            static fn(WorkerMessage $message) => $listener->workerMessage($handle, $message),
192            fn(string $reason) => $this->onWorkerGone($state, $handle, $reason),
193        );
194        $listener->workerSpawned($handle);
195
196        $this->logger->debug('Worker started', ['worker' => $handle->id->value, 'pid' => $handle->getPid(), 'apps' => $slot->listAppIds()->toStrings()]);
197    }
198
199    private function awaitWorkerExit(WorkerHandle $handle, Cancellation $deadline): void
200    {
201        try {
202            $summary = $handle->awaitExit($deadline);
203            $this->logger->debug('Worker stopped cleanly', ['worker' => $handle->id->value, 'summary' => $summary]);
204        } catch (Throwable $e) {
205            if (!$handle->isTerminated()) {
206                $this->logger->warning('Worker did not stop in time; killing it', [
207                    'worker' => $handle->id->value,
208                    'exception' => $e,
209                ]);
210            }
211        } finally {
212            $handle->terminate();
213        }
214    }
215
216    private function armReadyDeadline(WorkerHandle $handle): ?TimerHandle
217    {
218        $readyTimeout = $this->supervision->readyTimeout->findDuration();
219
220        if ($readyTimeout === null) {
221            return null;
222        }
223
224        // A worker that answers pings but never reports ready is treated as dead.
225        return $this->timers->startTimer($readyTimeout, function () use ($handle, $readyTimeout): void {
226            if (!$handle->isStarting()) {
227                return;
228            }
229
230            $this->logger->error('Worker never reported its apps ready; restarting it', [
231                'worker' => $handle->id->value,
232                'pid' => $handle->getPid(),
233                'apps' => $handle->getAppIds()->toStrings(),
234                'ready_timeout' => (string) $readyTimeout,
235            ]);
236
237            $handle->terminate();
238        });
239    }
240
241    private function onWorkerGone(WorkerSlotState $state, WorkerHandle $handle, string $reason): void
242    {
243        // Only the live handle counts: a displaced handle or a slot stopped by shutdown must not restart.
244        if (!$state->isLive($handle)) {
245            return;
246        }
247
248        if (!$handle->isTerminated()) {
249            $this->logUnexpectedExit($handle);
250        }
251
252        // Shutdown may have stopped the slot while the exit result was awaited.
253        if (!$state->isLive($handle)) {
254            return;
255        }
256
257        $handle->terminate();
258        $state->markGone();
259        $state->listener->workerGone($handle, $reason);
260
261        if ($this->running) {
262            $this->logger->error('Worker died', ['worker' => $handle->id->value, 'pid' => $handle->getPid(), 'apps' => $handle->getAppIds()->toStrings(), 'reason' => $reason]);
263            $this->scheduleRestart($state);
264        }
265    }
266
267    private function logUnexpectedExit(WorkerHandle $handle): void
268    {
269        try {
270            $summary = $handle->awaitExit($this->timers->timeout(Duration::seconds(self::EXIT_RESULT_WAIT_SECONDS)));
271            $this->logger->warning('Worker exited', ['worker' => $handle->id->value, 'summary' => $summary]);
272        } catch (CancelledException) {
273            return;
274        } catch (Throwable $e) {
275            $this->logger->error('Worker exited with an error', ['worker' => $handle->id->value, 'exception' => $e]);
276        }
277    }
278
279    private function scheduleRestart(WorkerSlotState $state): void
280    {
281        $slot = $state->slot;
282        $decision = $this->restartPolicy->decideRestart($slot->workerId);
283
284        if ($decision->delay === null) {
285            $this->logger->error('Worker quarantined after repeated failures; its automations are not running', [
286                'worker' => $slot->workerId->value,
287                'apps' => $slot->listAppIds()->toStrings(),
288                'attempts' => $this->supervision->restartAttempts,
289                'window' => (string) $this->supervision->restartWindow,
290            ]);
291
292            $state->markQuarantined();
293            $state->listener->workerQuarantined($slot);
294
295            return;
296        }
297
298        $delay = $decision->delay;
299
300        $this->logger->info('Restarting worker', ['worker' => $slot->workerId->value, 'apps' => $slot->listAppIds()->toStrings(), 'in' => (string) $delay]);
301
302        $state->recordScheduledRestart(
303            $this->timers->startTimer($delay, fn() => $this->respawnWorker($state)),
304            $this->clock->getNow()->plus($delay),
305        );
306    }
307
308    private function respawnOnRequest(WorkerSlotState $state, ?WorkerHandle $stopping): void
309    {
310        if ($stopping !== null) {
311            $stopping->sendShutdown(new Shutdown('restart requested', $this->config->shutdownGrace));
312            $this->awaitWorkerExit($stopping, $this->timers->timeout($this->config->shutdownGrace->plus(Duration::milliseconds(self::JOIN_MARGIN_MILLISECONDS))));
313            $state->listener->workerGone($stopping, 'restart requested');
314        }
315
316        $this->respawnWorker($state);
317    }
318
319    private function respawnWorker(WorkerSlotState $state): void
320    {
321        try {
322            $this->startWorker($state->slot, $state->listener);
323        } catch (Throwable $e) {
324            $this->logger->error('Could not restart worker', ['worker' => $state->slot->workerId->value, 'exception' => $e]);
325        }
326    }
327}