Code Coverage
 
Lines
Functions and Methods
Classes and Traits
Total
100.00%
73 / 73
100.00%
13 / 13
 CRAP
100.00%
1 / 1
RetryingIcapClient
100.00%
73 / 73
100.00%
13 / 13
 31
100.00%
1 / 1
 __construct
100.00%
12 / 12
100.00%
1 / 1
 10
 request
100.00%
4 / 4
100.00%
1 / 1
 1
 options
100.00%
12 / 12
100.00%
1 / 1
 4
 scanFile
100.00%
4 / 4
100.00%
1 / 1
 1
 scanFileWithPreview
100.00%
4 / 4
100.00%
1 / 1
 1
 scanFileDiscardingBody
100.00%
5 / 5
100.00%
1 / 1
 1
 scanFileWithPreviewDiscardingBody
100.00%
5 / 5
100.00%
1 / 1
 1
 discardingInner
100.00%
3 / 3
100.00%
1 / 1
 2
 once
100.00%
2 / 2
100.00%
1 / 1
 1
 run
100.00%
10 / 10
100.00%
1 / 1
 2
 pause
100.00%
7 / 7
100.00%
1 / 1
 3
 pauseAfter
100.00%
2 / 2
100.00%
1 / 1
 2
 backoffCap
100.00%
3 / 3
100.00%
1 / 1
 2
1<?php
2
3/**
4 * SPDX-License-Identifier: EUPL-1.2
5 *
6 * This file is part of icap-flow.
7 *
8 * Licensed under the EUPL, Version 1.2 only (the "Licence");
9 * you may not use this work except in compliance with the Licence.
10 * You may obtain a copy of the Licence at:
11 *
12 *     https://joinup.ec.europa.eu/collection/eupl/eupl-text-eupl-12
13 *
14 * Unless required by applicable law or agreed to in writing, software
15 * distributed under the Licence is distributed on an "AS IS" basis,
16 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
17 */
18
19declare(strict_types=1);
20
21namespace Ndrstmr\Icap;
22
23use Amp\Cancellation;
24use Amp\CancelledException;
25use Amp\Future;
26use Closure;
27use Ndrstmr\Icap\DTO\IcapRequest;
28use Ndrstmr\Icap\DTO\IcapResponse;
29use Ndrstmr\Icap\DTO\ScanResult;
30use Ndrstmr\Icap\Exception\IcapServerException;
31
32use function Amp\async;
33use function Amp\delay;
34
35/**
36 * Retry decorator around any {@see IcapClientInterface}.
37 *
38 * **What is repeated (N03).** Only {@see options()}, and only after an
39 * {@see IcapServerException} (ICAP 5xx), which is transient by RFC 3507
40 * §4.3.3: OPTIONS has no body and changes nothing, so a second one is safe.
41 * {@see request()}, {@see scanFile()}, {@see scanFileWithPreview()} and their discarding
42 * variants get one attempt. A 5xx, or a write that failed, does not prove that the server did
43 * not process any of the bytes it was sent, and nothing here can send an upload
44 * again (a stream that was read cannot be read again, and a file may have
45 * changed); an application that knows its scans are repeatable repeats them
46 * itself. Client errors (4xx), parse failures, connection errors and timeouts
47 * propagate at once: retrying them would be useless or would add to a
48 * hostile environment.
49 *
50 * **Backoff.** Exponential with full jitter: the pause before the next attempt
51 * is a random share, from nothing up to all of it, of
52 * `baseDelaySeconds * backoffFactor^(attempt-1)`, capped at `maxDelaySeconds`,
53 * so that many clients do not retry in step. With the defaults (100 ms / 2x /
54 * cap 5 s) the pauses are at most 0.1 / 0.2 / 0.4 s. The pause ends with the
55 * caller's cancellation and with the end of the operation, and even a pause of
56 * 0 seconds hands the event loop over once.
57 *
58 * **One budget (M10).** `operationTimeout` (60 seconds by default) is the whole
59 * of a call, attempts and pauses together, on a monotonic clock. It begins when
60 * the method is called, is not restarted by an attempt, and is handed to the
61 * inner client as its cancellation: an inner {@see IcapClient} ends with it (one
62 * with a shorter timeout of its own keeps it; none can extend this one). It ends
63 * the call with an {@see \Ndrstmr\Icap\Exception\IcapTimeoutException}; the
64 * caller's cancellation stays an `Amp\CancelledException`. An inner client that
65 * ignores its cancellation is waited for, and a result that comes in after the
66 * end is not returned.
67 *
68 * Example wiring:
69 *
70 *     $client = new RetryingIcapClient(
71 *         IcapClient::create(),
72 *         maxAttempts: 3,
73 *         baseDelaySeconds: 0.1,
74 *         maxDelaySeconds: 5.0,
75 *         operationTimeout: 30.0,
76 *     );
77 */
78final class RetryingIcapClient implements IcapClientInterface, DiscardingIcapClient
79{
80    /** @var (Closure(float): void)|null */
81    private ?Closure $sleeper;
82
83    /** @var Closure(): float */
84    private Closure $random;
85
86    /**
87     * @param IcapClientInterface           $inner            Wrapped client (typically IcapClient)
88     * @param int                           $maxAttempts      Maximum total attempts (initial + retries); must be >= 1
89     * @param float                         $baseDelaySeconds Initial delay before the first retry; finite, 0 or more
90     * @param float                         $maxDelaySeconds  Cap on any individual retry delay; finite, 0 or more
91     * @param float                         $backoffFactor    Multiplier applied between attempts; finite and above 0
92     * @param (Closure(float): void)|null   $sleeper          Test seam — replaces the pause with a callable. It is not interrupted
93     *                                                        from outside; the end of the operation is looked at before and after it
94     * @param float                         $operationTimeout Seconds one call may take in all, attempts and pauses together;
95     *                                                        finite and above 0
96     * @param (Closure(): float)|null       $random           Test seam — the random share of the jitter, from 0 to 1 (default: mt_rand)
97     */
98    public function __construct(
99        private IcapClientInterface $inner,
100        private int $maxAttempts = 3,
101        private float $baseDelaySeconds = 0.1,
102        private float $maxDelaySeconds = 5.0,
103        private float $backoffFactor = 2.0,
104        ?Closure $sleeper = null,
105        private float $operationTimeout = 60.0,
106        ?Closure $random = null,
107    ) {
108        if ($maxAttempts < 1) {
109            throw new \InvalidArgumentException('maxAttempts must be >= 1, got: ' . $maxAttempts);
110        }
111        // var_export() for the floats: PHP 8.5 warns when a NAN is coerced to a string. NAN and INF pass a
112        // comparison with 0 and would turn the backoff into NAN, so finiteness is checked first.
113        if (!is_finite($baseDelaySeconds) || $baseDelaySeconds < 0.0) {
114            throw new \InvalidArgumentException('baseDelaySeconds must be a finite number >= 0, got: ' . var_export($baseDelaySeconds, true));
115        }
116        if (!is_finite($maxDelaySeconds) || $maxDelaySeconds < 0.0) {
117            throw new \InvalidArgumentException('maxDelaySeconds must be a finite number >= 0, got: ' . var_export($maxDelaySeconds, true));
118        }
119        if (!is_finite($backoffFactor) || $backoffFactor <= 0.0) {
120            throw new \InvalidArgumentException('backoffFactor must be a finite number > 0, got: ' . var_export($backoffFactor, true));
121        }
122        if (!is_finite($operationTimeout) || $operationTimeout <= 0.0) {
123            throw new \InvalidArgumentException('operationTimeout must be a finite number of seconds above 0, got: ' . var_export($operationTimeout, true));
124        }
125
126        $this->sleeper = $sleeper;
127        $this->random = $random ?? static fn (): float => mt_rand() / mt_getrandmax();
128    }
129
130    #[\Override]
131    public function request(IcapRequest $request, ?Cancellation $cancellation = null): Future
132    {
133        return $this->once(
134            $cancellation,
135            fn (OperationDeadline $deadline): ScanResult => $this->inner->request($request, $deadline)->await(),
136        );
137    }
138
139    /**
140     * The one operation that is repeated: after an ICAP 5xx, up to `maxAttempts` attempts in all, inside one
141     * `operationTimeout`.
142     */
143    #[\Override]
144    public function options(string $service, ?Cancellation $cancellation = null): Future
145    {
146        $deadline = OperationDeadline::begin($cancellation, $this->operationTimeout);
147
148        return $this->run($deadline, function () use ($service, $deadline): IcapResponse {
149            $last = null;
150            for ($attempt = 1; ; $attempt++) {
151                // No new attempt once the budget is gone, whatever the pause before it did.
152                $deadline->checkpoint($last);
153
154                try {
155                    return $this->inner->options($service, $deadline)->await();
156                } catch (IcapServerException $e) {
157                    $last = $e;
158                    if ($attempt >= $this->maxAttempts) {
159                        throw $e;
160                    }
161                    $this->pause($this->pauseAfter($attempt), $deadline, $e);
162                }
163            }
164        });
165    }
166
167    /**
168     * @param array<string, string|string[]> $extraHeaders
169     */
170    #[\Override]
171    public function scanFile(
172        string $service,
173        string $filePath,
174        array $extraHeaders = [],
175        ?Cancellation $cancellation = null,
176    ): Future {
177        return $this->once(
178            $cancellation,
179            fn (OperationDeadline $deadline): ScanResult => $this->inner->scanFile($service, $filePath, $extraHeaders, $deadline)->await(),
180        );
181    }
182
183    /**
184     * @param array<string, string|string[]> $extraHeaders
185     */
186    #[\Override]
187    public function scanFileWithPreview(
188        string $service,
189        string $filePath,
190        ?int $previewSize = null,
191        array $extraHeaders = [],
192        ?Cancellation $cancellation = null,
193    ): Future {
194        return $this->once(
195            $cancellation,
196            fn (OperationDeadline $deadline): ScanResult => $this->inner->scanFileWithPreview($service, $filePath, $previewSize, $extraHeaders, $deadline)->await(),
197        );
198    }
199
200    /**
201     * @param array<string, string|string[]> $extraHeaders
202     *
203     * @throws \LogicException When the wrapped client does not implement {@see DiscardingIcapClient}
204     */
205    #[\Override]
206    public function scanFileDiscardingBody(
207        string $service,
208        string $filePath,
209        int $maxResponseBytes,
210        array $extraHeaders = [],
211        ?Cancellation $cancellation = null,
212    ): Future {
213        $inner = $this->discardingInner();
214
215        return $this->once(
216            $cancellation,
217            fn (OperationDeadline $deadline): ScanResult => $inner->scanFileDiscardingBody($service, $filePath, $maxResponseBytes, $extraHeaders, $deadline)->await(),
218        );
219    }
220
221    /**
222     * @param array<string, string|string[]> $extraHeaders
223     *
224     * @throws \LogicException When the wrapped client does not implement {@see DiscardingIcapClient}
225     */
226    #[\Override]
227    public function scanFileWithPreviewDiscardingBody(
228        string $service,
229        string $filePath,
230        int $maxResponseBytes,
231        ?int $previewSize = null,
232        array $extraHeaders = [],
233        ?Cancellation $cancellation = null,
234    ): Future {
235        $inner = $this->discardingInner();
236
237        return $this->once(
238            $cancellation,
239            fn (OperationDeadline $deadline): ScanResult => $inner->scanFileWithPreviewDiscardingBody($service, $filePath, $maxResponseBytes, $previewSize, $extraHeaders, $deadline)->await(),
240        );
241    }
242
243    private function discardingInner(): DiscardingIcapClient
244    {
245        return $this->inner instanceof DiscardingIcapClient
246            ? $this->inner
247            : throw new \LogicException($this->inner::class . ' cannot discard the body of a response; it does not implement ' . DiscardingIcapClient::class . '.');
248    }
249
250    /**
251     * One attempt, inside the budget of the call.
252     *
253     * @template T
254     *
255     * @param Closure(OperationDeadline): T $operation
256     *
257     * @return Future<T>
258     */
259    private function once(?Cancellation $cancellation, Closure $operation): Future
260    {
261        $deadline = OperationDeadline::begin($cancellation, $this->operationTimeout);
262
263        return $this->run($deadline, static fn (): mixed => $operation($deadline));
264    }
265
266    /**
267     * Runs the operation in a fiber of its own, with the timer and the subscription of the budget set for exactly
268     * as long as it runs. A result that comes in after the end of the budget is none.
269     *
270     * @template T
271     *
272     * @param Closure(): T $operation
273     *
274     * @return Future<T>
275     */
276    private function run(OperationDeadline $deadline, Closure $operation): Future
277    {
278        /** @var Future<T> $future */
279        $future = async(static function () use ($deadline, $operation): mixed {
280            try {
281                $deadline->checkpoint();
282                $result = $operation();
283                $deadline->checkpoint();
284
285                return $result;
286            } catch (\Throwable $e) {
287                throw $deadline->outcome($e);
288            } finally {
289                $deadline->close();
290            }
291        });
292
293        return $future;
294    }
295
296    /**
297     * The pause between two attempts. It ends with the caller's cancellation and with the end of the operation (an
298     * {@see \Ndrstmr\Icap\Exception\IcapTimeoutException} with the 5xx as its cause). A sleeper cannot be interrupted
299     * from here, so the budget is looked at before it is called and, by the loop, before the next attempt.
300     */
301    private function pause(float $seconds, OperationDeadline $deadline, IcapServerException $cause): void
302    {
303        if ($this->sleeper !== null) {
304            $deadline->checkpoint($cause);
305            ($this->sleeper)($seconds);
306
307            return;
308        }
309
310        try {
311            // Also a pause of 0 seconds suspends once: a loop of attempts must leave the event loop its turns.
312            delay($seconds, true, $deadline);
313        } catch (CancelledException $e) {
314            throw $deadline->outcome($e, $cause);
315        }
316    }
317
318    /**
319     * Full jitter: any pause from nothing up to the capped exponential one.
320     */
321    private function pauseAfter(int $attempt): float
322    {
323        $draw = ($this->random)();
324
325        return $this->backoffCap($attempt) * (is_finite($draw) ? min(max($draw, 0.0), 1.0) : 1.0);
326    }
327
328    /**
329     * `baseDelaySeconds * backoffFactor^(attempt-1)`, capped at `maxDelaySeconds`. The power overflows to INF for
330     * a large attempt number, and INF is above every cap; a base of 0 is a pause of 0 before the product with INF
331     * could become NAN.
332     */
333    private function backoffCap(int $attempt): float
334    {
335        if ($this->baseDelaySeconds === 0.0) {
336            return 0.0;
337        }
338
339        return min($this->baseDelaySeconds * ($this->backoffFactor ** ($attempt - 1)), $this->maxDelaySeconds);
340    }
341}