Lines 89.09% 49 / 55
Functions and Methods 90.00% 9 / 10
Classes and Traits 0.00% 0 / 1
Name Lines Functions and Methods CRAP Classes and Traits
HaCallSlots 89.09% 49 / 55 90.00% 9 / 10 22.63 0.00% 0 / 1
 __construct 100.00% 2 / 2 100.00% 1 / 1 1
 runBounded 100.00% 5 / 5 100.00% 1 / 1 2
 runCounted 100.00% 1 / 1 100.00% 1 / 1 1
 countInFlightCallsFor 100.00% 1 / 1 100.00% 1 / 1 1
 runInSlot 70.00% 14 / 20 0.00% 0 / 1 4.43
 describeFailure 100.00% 2 / 2 100.00% 1 / 1 2
 findRefusalReason 100.00% 5 / 5 100.00% 1 / 1 5
 refuse 100.00% 11 / 11 100.00% 1 / 1 3
 acquire 100.00% 2 / 2 100.00% 1 / 1 1
 release 100.00% 6 / 6 100.00% 1 / 1 2
1<?php
2
3declare(strict_types=1);
4
5namespace Stewart\Runtime\Broker;
6
7use Closure;
8use Psr\Log\LoggerInterface;
9use Stewart\Contracts\Exception\ExceptionReason;
10use Stewart\Contracts\Exception\StewartException;
11use Stewart\Contracts\Time\Clock;
12use Stewart\Runtime\Config\ServiceCallPolicy;
13use Stewart\Runtime\Ipc\Message\BrokerMessage;
14use Stewart\Runtime\Ipc\Message\HaCallRequest;
15use Stewart\Runtime\Ipc\Message\RefusableHaCallRequest;
16use Stewart\Runtime\Ipc\Message\RequestFailed;
17use Stewart\Runtime\Logging\EveryNthOccurrence;
18use Stewart\Runtime\Model\ServiceCallOutcome;
19use Throwable;
20use WeakMap;
21
22use function Amp\async;
23
24final class HaCallSlots
25{
26    /** @var WeakMap<WorkerHandle, int> */
27    private WeakMap $perWorker;
28
29    public private(set) int $inFlight = 0;
30
31    public private(set) int $refusedCalls = 0;
32
33    private readonly EveryNthOccurrence $refusalReports;
34
35    public function __construct(
36        private readonly ServiceCallPolicy $policy,
37        private readonly AppMetrics $metrics,
38        private readonly Clock $clock,
39        private readonly LoggerInterface $logger,
40    ) {
41        $this->perWorker = new WeakMap();
42        $this->refusalReports = EveryNthOccurrence::forRepeatedWarnings();
43    }
44
45    /** @param Closure(): BrokerMessage $call */
46    public function runBounded(WorkerHandle $handle, RefusableHaCallRequest $request, Closure $call, bool $metered = true): void
47    {
48        $refusal = $this->findRefusalReason($handle);
49
50        if ($refusal !== null) {
51            $this->refuse($handle, $request, $refusal, $metered);
52
53            return;
54        }
55
56        $this->runInSlot($handle, $request, $call, $metered);
57    }
58
59    // Exposure updates hold a slot but are never refused, so HA never keeps a stale exposed state.
60    /** @param Closure(): BrokerMessage $call */
61    public function runCounted(WorkerHandle $handle, HaCallRequest $request, Closure $call): void
62    {
63        $this->runInSlot($handle, $request, $call, metered: false);
64    }
65
66    public function countInFlightCallsFor(WorkerHandle $handle): int
67    {
68        return $this->perWorker[$handle] ?? 0;
69    }
70
71    /** @param Closure(): BrokerMessage $call */
72    private function runInSlot(WorkerHandle $handle, HaCallRequest $request, Closure $call, bool $metered): void
73    {
74        $this->acquire($handle);
75        $startedAt = $this->clock->getMonotonicTime();
76
77        async(function () use ($handle, $request, $call, $metered, $startedAt): void {
78            try {
79                try {
80                    $reply = $call();
81                    $outcome = ServiceCallOutcome::Succeeded;
82                } catch (Throwable $e) {
83                    $failure = $this->describeFailure($request, $e);
84                    $reply = RequestFailed::fromException($request->correlationId, $failure);
85                    $outcome = ServiceCallOutcome::failedWith($failure->reason);
86                } finally {
87                    $this->release($handle);
88                }
89
90                $handle->send($reply);
91
92                if ($metered) {
93                    $this->metrics->recordFinishedCall($handle->id, $request->scope, $outcome, $this->clock->getMonotonicTime()->elapsedSince($startedAt));
94                }
95            } catch (Throwable $e) {
96                $this->logger->error('Home Assistant call reply failed', [
97                    'exception' => $e,
98                    'worker' => $handle->id->value,
99                    'correlation_id' => $request->correlationId->value,
100                ]);
101            }
102        })->ignore();
103    }
104
105    // A failure of the request's own kind passes through; anything else means Home Assistant was not reached.
106    /** @return StewartException<ExceptionReason> */
107    private function describeFailure(HaCallRequest $request, Throwable $e): StewartException
108    {
109        $unreachable = $request->createUnreachableFailure($e->getMessage(), $e);
110
111        return $e instanceof $unreachable ? $e : $unreachable;
112    }
113
114    private function findRefusalReason(WorkerHandle $handle): ?string
115    {
116        if ($this->policy->boundsTotal() && $this->inFlight >= $this->policy->total) {
117            return \sprintf('the broker already has %d calls to Home Assistant in flight', $this->policy->total);
118        }
119
120        if ($this->policy->limitsPerWorker() && $this->countInFlightCallsFor($handle) >= $this->policy->perWorker) {
121            return \sprintf('this worker already has %d calls to Home Assistant in flight', $this->policy->perWorker);
122        }
123
124        return null;
125    }
126
127    private function refuse(WorkerHandle $handle, RefusableHaCallRequest $request, string $reason, bool $metered): void
128    {
129        ++$this->refusedCalls;
130
131        if ($this->refusalReports->includesOccurrence($this->refusedCalls)) {
132            $this->logger->warning('Refusing calls; Home Assistant is not keeping up', [
133                'worker' => $handle->id->value,
134                'app' => $request->scope->wireValue(),
135                'refused' => $this->refusedCalls,
136                'in_flight' => $this->inFlight,
137            ]);
138        }
139
140        if ($metered) {
141            $this->metrics->recordRefusedCall($handle->id, $request->scope);
142        }
143
144        $handle->send(RequestFailed::fromException($request->correlationId, $request->createOverloadedFailure($reason)));
145    }
146
147    private function acquire(WorkerHandle $handle): void
148    {
149        ++$this->inFlight;
150        $this->perWorker[$handle] = $this->countInFlightCallsFor($handle) + 1;
151    }
152
153    private function release(WorkerHandle $handle): void
154    {
155        --$this->inFlight;
156
157        $forWorker = $this->countInFlightCallsFor($handle) - 1;
158
159        if ($forWorker <= 0) {
160            unset($this->perWorker[$handle]);
161
162            return;
163        }
164
165        $this->perWorker[$handle] = $forWorker;
166    }
167}