Lines 80.76% 63 / 78
Functions and Methods 79.16% 19 / 24
Classes and Traits 0.00% 0 / 1
Name Lines Functions and Methods CRAP Classes and Traits
WorkerHandle 80.76% 63 / 78 79.16% 19 / 24 51.38 0.00% 0 / 1
 __construct 100.00% 3 / 3 100.00% 1 / 1 1
 getPid 100.00% 1 / 1 100.00% 1 / 1 1
 getAppIds 100.00% 1 / 1 100.00% 1 / 1 1
 getTransport 100.00% 1 / 1 100.00% 1 / 1 1
 relayOutput 100.00% 2 / 2 100.00% 1 / 1 1
 isReady 100.00% 1 / 1 100.00% 1 / 1 1
 isStarting 100.00% 1 / 1 100.00% 1 / 1 1
 isStopping 100.00% 1 / 1 100.00% 1 / 1 1
 isTerminated 100.00% 1 / 1 100.00% 1 / 1 1
 send 100.00% 5 / 5 100.00% 1 / 1 2
 sendStateChange 0.00% 0 / 4 0.00% 0 / 1 6
 sendShutdown 100.00% 5 / 5 100.00% 1 / 1 2
 markReady 100.00% 1 / 1 100.00% 1 / 1 1
 recordPong 100.00% 3 / 3 100.00% 1 / 1 1
 countMissedProbes 100.00% 1 / 1 100.00% 1 / 1 1
 getOutboxStatus 100.00% 1 / 1 100.00% 1 / 1 1
 awaitExit 100.00% 3 / 3 100.00% 1 / 1 1
 terminate 100.00% 5 / 5 100.00% 1 / 1 2
 acceptsMessages 100.00% 1 / 1 100.00% 1 / 1 2
 moveTo 66.66% 2 / 3 0.00% 0 / 1 2.15
 relayLines 71.42% 5 / 7 0.00% 0 / 1 4.37
 handleFailedSend 85.71% 6 / 7 0.00% 0 / 1 3.03
 reportDrops 30.00% 3 / 10 0.00% 0 / 1 6.09
 flush 100.00% 10 / 10 100.00% 1 / 1 4
1<?php
2
3declare(strict_types=1);
4
5namespace Stewart\Runtime\Broker;
6
7use Amp\ByteStream;
8use Amp\ByteStream\ReadableStream;
9use Amp\Cancellation;
10use Amp\Future;
11use LogicException;
12use Psr\Log\LoggerInterface;
13use Stewart\Contracts\App\Collection\AppIdCollection;
14use Stewart\Contracts\Time\Instant;
15use Stewart\Runtime\Control\Protocol\Status\OutboxStatus;
16use Stewart\Runtime\Ipc\Message\BrokerMessage;
17use Stewart\Runtime\Ipc\Message\Pong;
18use Stewart\Runtime\Ipc\Message\Shutdown;
19use Stewart\Runtime\Ipc\Transport;
20use Stewart\Runtime\Ipc\Wire\EncodedStateChange;
21use Stewart\Runtime\Lifecycle\WorkerHandleState;
22use Stewart\Runtime\Logging\EveryNthOccurrence;
23use Stewart\Runtime\Model\WorkerId;
24use Throwable;
25
26use function Amp\async;
27
28final class WorkerHandle
29{
30    private readonly WorkerOutbox $outbox;
31
32    private readonly EveryNthOccurrence $dropReports;
33
34    private bool $flushing = false;
35
36    private WorkerHandleState $state = WorkerHandleState::Starting;
37
38    public private(set) ?Pong $lastPong = null;
39
40    public private(set) ?Instant $lastPongAt = null;
41
42    private int $lastAnsweredNonce;
43
44    private int $reportedDrops = 0;
45
46    public function __construct(
47        public readonly WorkerId $id,
48        private readonly WorkerProcess $process,
49        public readonly WorkerSlot $slot,
50        private readonly LoggerInterface $logger,
51        OutboxLimits $outboxLimits,
52        int $probesSentBefore = 0,
53    ) {
54        $this->outbox = new WorkerOutbox($outboxLimits);
55        $this->dropReports = EveryNthOccurrence::forRepeatedWarnings();
56        // Probes sent before the spawn do not count as missed.
57        $this->lastAnsweredNonce = $probesSentBefore;
58    }
59
60    public function getPid(): int
61    {
62        return $this->process->getPid();
63    }
64
65    public function getAppIds(): AppIdCollection
66    {
67        return $this->slot->listAppIds();
68    }
69
70    public function getTransport(): Transport
71    {
72        return $this->process->transport();
73    }
74
75    public function relayOutput(): void
76    {
77        $this->relayLines($this->process->stdout(), 'stdout');
78        $this->relayLines($this->process->stderr(), 'stderr');
79    }
80
81    public function isReady(): bool
82    {
83        return $this->state === WorkerHandleState::Ready;
84    }
85
86    public function isStarting(): bool
87    {
88        return $this->state === WorkerHandleState::Starting;
89    }
90
91    public function isStopping(): bool
92    {
93        return $this->state === WorkerHandleState::Stopping;
94    }
95
96    public function isTerminated(): bool
97    {
98        return $this->state === WorkerHandleState::Terminated;
99    }
100
101    public function send(BrokerMessage $message): void
102    {
103        if (!$this->acceptsMessages()) {
104            return;
105        }
106
107        $this->outbox->enqueue($message);
108        $this->reportDrops();
109        $this->flush();
110    }
111
112    public function sendStateChange(EncodedStateChange $change): void
113    {
114        if (!$this->acceptsMessages()) {
115            return;
116        }
117
118        $this->outbox->pushStateChange($change);
119        $this->flush();
120    }
121
122    public function sendShutdown(Shutdown $shutdown): void
123    {
124        if (!$this->acceptsMessages()) {
125            return;
126        }
127
128        $this->outbox->enqueue($shutdown);
129        $this->moveTo(WorkerHandleState::Stopping);
130        $this->flush();
131    }
132
133    public function markReady(): void
134    {
135        $this->moveTo(WorkerHandleState::Ready);
136    }
137
138    public function recordPong(Pong $pong, Instant $at): void
139    {
140        $this->lastAnsweredNonce = max($this->lastAnsweredNonce, $pong->nonce);
141        $this->lastPong = $pong;
142        $this->lastPongAt = $at;
143    }
144
145    public function countMissedProbes(int $nextNonce): int
146    {
147        return max(0, $nextNonce - $this->lastAnsweredNonce - 1);
148    }
149
150    public function getOutboxStatus(): OutboxStatus
151    {
152        return $this->outbox->buildStatus();
153    }
154
155    // amphp's join waits for the process exit without a deadline; terminate() ends the abandoned wait.
156    public function awaitExit(Cancellation $deadline): string
157    {
158        /** @var Future<string> $exit */
159        $exit = async(fn(): string => $this->process->join());
160        $exit->ignore();
161
162        return $exit->await($deadline);
163    }
164
165    public function terminate(): void
166    {
167        if ($this->isTerminated()) {
168            return;
169        }
170
171        $this->moveTo(WorkerHandleState::Terminated);
172        $this->outbox->clear();
173        $this->process->close();
174    }
175
176    private function acceptsMessages(): bool
177    {
178        return $this->state === WorkerHandleState::Starting || $this->state === WorkerHandleState::Ready;
179    }
180
181    private function moveTo(WorkerHandleState $state): void
182    {
183        if (!$this->state->canEnter($state)) {
184            throw new LogicException(\sprintf('Worker %d handle cannot move from %s to %s.', $this->id->value, $this->state->name, $state->name));
185        }
186
187        $this->state = $state;
188    }
189
190    private function relayLines(ReadableStream $stream, string $name): void
191    {
192        async(function () use ($stream, $name): void {
193            try {
194                foreach (ByteStream\splitLines($stream) as $line) {
195                    if ($line !== '') {
196                        $this->logger->notice($line, ['worker' => $this->id->value, 'stream' => $name]);
197                    }
198                }
199            } catch (Throwable $e) {
200                $this->logger->warning('Lost a worker output stream', ['worker' => $this->id->value, 'stream' => $name, 'exception' => $e]);
201            }
202        })->ignore();
203    }
204
205    private function handleFailedSend(Throwable $failure): void
206    {
207        if ($this->isTerminated()) {
208            return;
209        }
210
211        // A stopping worker closes its channel before exiting; the shutdown deadline still bounds the wait.
212        if ($this->isStopping()) {
213            $this->logger->debug('Stopped sending to a stopping worker', ['worker' => $this->id->value, 'reason' => $failure->getMessage()]);
214
215            return;
216        }
217
218        $this->logger->warning('Failed sending to worker; terminating it', ['worker' => $this->id->value, 'exception' => $failure]);
219        $this->terminate();
220    }
221
222    private function reportDrops(): void
223    {
224        $dropped = $this->outbox->dropped;
225
226        if ($dropped === $this->reportedDrops) {
227            return;
228        }
229
230        $this->reportedDrops = $dropped;
231
232        if ($this->dropReports->includesOccurrence($dropped)) {
233            $this->logger->warning('Worker is falling behind; dropping its oldest events and topic messages', [
234                'worker' => $this->id->value,
235                'dropped' => $dropped,
236                'coalesced_state_changes' => $this->outbox->coalescedStateChanges,
237            ]);
238        }
239    }
240
241    private function flush(): void
242    {
243        if ($this->flushing) {
244            return;
245        }
246
247        $this->flushing = true;
248
249        async(function (): void {
250            try {
251                while (($message = $this->outbox->takeNext()) !== null) {
252                    $this->getTransport()->send($message);
253                }
254            } catch (Throwable $e) {
255                $this->handleFailedSend($e);
256            } finally {
257                $this->flushing = false;
258            }
259        })->ignore();
260    }
261}