Code Coverage
 
Lines
Functions and Methods
Classes and Traits
Total
100.00%
110 / 110
100.00%
10 / 10
 CRAP
100.00%
1 / 1
AmpTransportSession
100.00%
110 / 110
100.00%
10 / 10
 29
100.00%
1 / 1
 __construct
100.00%
5 / 5
100.00%
1 / 1
 1
 write
100.00%
16 / 16
100.00%
1 / 1
 4
 readResponse
100.00%
1 / 1
100.00%
1 / 1
 1
 readResponseDiscardingBody
100.00%
1 / 1
100.00%
1 / 1
 1
 receive
100.00%
25 / 25
100.00%
1 / 1
 4
 inWindow
100.00%
47 / 47
100.00%
1 / 1
 9
 socketFailed
100.00%
2 / 2
100.00%
1 / 1
 1
 release
100.00%
7 / 7
100.00%
1 / 1
 4
 close
100.00%
4 / 4
100.00%
1 / 1
 2
 assertActive
100.00%
2 / 2
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\Transport;
22
23use Amp\ByteStream\StreamException;
24use Amp\Cancellation;
25use Amp\CancelledException;
26use Amp\DeferredCancellation;
27use Amp\Socket\Socket as SocketInterface;
28use Ndrstmr\Icap\Config;
29use Ndrstmr\Icap\DTO\IcapResponse;
30use Ndrstmr\Icap\Exception\IcapConnectionException;
31use Ndrstmr\Icap\Exception\IcapTimeoutException;
32use Revolt\EventLoop;
33
34/**
35 * One-acquired-socket binding used for multi-round-trip flows like
36 * the strict RFC 3507 §4.5 preview-continue exchange.
37 *
38 * Lifecycle:
39 *   - {@see AsyncAmpTransport::openSession()} acquires a socket from
40 *     the pool (or opens a fresh one).
41 *   - The caller writes / reads as many times as the protocol
42 *     requires.
43 *   - The caller calls {@see release()} on success (returns to the
44 *     pool when one is configured) or {@see close()} on failure.
45 *
46 * Every {@see write()} call gets a window of `streamTimeout` seconds of its own,
47 * shared by everything that call does — not one window per chunk, and no timer
48 * that spans the session. A {@see readResponse()} call has two, one after the
49 * other (M10): `responseStartTimeout` for the server to begin its answer (the
50 * time a scan needs), then `streamTimeout` from the first byte for the answer
51 * to arrive in full, a maximum that no byte starts again. A call also ends with
52 * the caller's cancellation, which in an operation of the client is also the end
53 * of the operation ({@see \Ndrstmr\Icap\OperationDeadline}). A call that ends in
54 * either, or in any other failure, closes the session for good: a request
55 * sent in part or a response read in part cannot be continued. {@see release()}
56 * does not hand the socket to a pool once the caller has cancelled or the
57 * operation has run out of time either. The window passing is an
58 * {@see IcapTimeoutException}; the caller's cancellation — also a
59 * `TimeoutCancellation` — stays an `Amp\CancelledException`.
60 *
61 * The producer of a request is part of the call. The event loop cannot fire
62 * the timer of the window while a producer holds it, so the window also has an
63 * end on the monotonic clock, and {@see write()} looks at the caller, the
64 * window and that end before it asks the producer for anything and wherever
65 * the producer hands control back: with every chunk (an empty one too), after
66 * every write and before it returns. A producer that never hands control back
67 * cannot be interrupted from outside.
68 *
69 * A read that brings in bytes behind the end of a response fails, and the session
70 * is closed: a connection carries one outstanding request at a time (RFC 3507 §4.1),
71 * so those bytes answer nothing, and the next user of the socket must not read them
72 * as its answer ({@see ResponseFrameReader}).
73 */
74final class AmpTransportSession implements DiscardingTransportSession
75{
76    private bool $disposed = false;
77
78    private readonly ResponseFrameReader $reader;
79
80    public function __construct(
81        private readonly Config $config,
82        private readonly SocketInterface $socket,
83        private readonly float $streamTimeout,
84        private readonly ?Cancellation $userCancellation,
85        int $maxResponseSize,
86        int $maxHeaderLineLength,
87        private readonly ?ConnectionPoolInterface $pool,
88    ) {
89        $this->reader = new ResponseFrameReader(
90            maxResponseSize: $maxResponseSize,
91            maxHeaderLineLength: $maxHeaderLineLength,
92            maxHeaderCount: $config->getMaxHeaderCount(),
93        );
94    }
95
96    #[\Override]
97    public function write(iterable $chunks): void
98    {
99        $this->assertActive();
100
101        $this->inWindow('Writing the request', 'stream timeout', $this->streamTimeout, function (Cancellation $window, \Closure $checkpoint) use ($chunks): void {
102            // Cancelled or out of time before it began: the producer is not asked for anything.
103            $checkpoint();
104
105            // Socket::write() takes no cancellation: against a peer that does not read it
106            // suspends this fiber until the peer does. Closing the socket is what fails the
107            // pending write, so the end of the window, by deadline or by the caller, closes it.
108            $watch = $window->subscribe(function (): void {
109                $this->close();
110            });
111
112            try {
113                foreach ($chunks as $chunk) {
114                    // The producer has handed control back; an empty chunk is no exception, no
115                    // write would fail for it.
116                    $checkpoint();
117                    if ($chunk !== '') {
118                        try {
119                            $this->socket->write($chunk);
120                        } catch (StreamException $e) {
121                            throw self::socketFailed($e, $checkpoint);
122                        }
123                    }
124                    // Before the producer is asked for the next chunk.
125                    $checkpoint();
126                }
127
128                // The producer may have run on after the last chunk went out.
129                $checkpoint();
130            } finally {
131                $window->unsubscribe($watch);
132            }
133        });
134    }
135
136    #[\Override]
137    public function readResponse(): string
138    {
139        return $this->receive(static fn (ResponseFrameReader $reader, \Closure $produce): string => $reader->readFrom($produce));
140    }
141
142    #[\Override]
143    public function readResponseDiscardingBody(int $maxResponseBytes): IcapResponse
144    {
145        return $this->receive(static fn (ResponseFrameReader $reader, \Closure $produce): IcapResponse => $reader->readDiscardingBody($produce, $maxResponseBytes));
146    }
147
148    /**
149     * One response, in the two windows of the read (M10), however the reader is to take it.
150     *
151     * @template T
152     *
153     * @param \Closure(ResponseFrameReader, \Closure(): ?string): T $consume
154     *
155     * @return T
156     */
157    private function receive(\Closure $consume): mixed
158    {
159        $this->assertActive();
160        $reader = $this->reader;
161        $socket = $this->socket;
162        $streamTimeout = $this->streamTimeout;
163        $begun = false;
164
165        return $this->inWindow(
166            'Waiting for the response to begin',
167            'response start timeout',
168            $this->config->getResponseStartTimeout(),
169            static function (Cancellation $window, \Closure $checkpoint, \Closure $next) use ($consume, $reader, $socket, $streamTimeout, &$begun): mixed {
170                $response = $consume($reader, static function () use ($socket, $window, $checkpoint, $next, $streamTimeout, &$begun): ?string {
171                    // No read is begun for a caller who has cancelled, nor once the stretch has run out.
172                    $checkpoint();
173                    try {
174                        $bytes = $socket->read($window);
175                    } catch (StreamException $e) {
176                        throw self::socketFailed($e, $checkpoint);
177                    }
178                    // The read may have found its bytes without waiting, or been held up by something that holds the
179                    // event loop. Bytes or the end of the stream count only if the caller has not cancelled and the
180                    // stretch the read began in has not run out (for the first bytes: the time to begin): otherwise the
181                    // decoder gets neither, and the end of the stream would be a framing error.
182                    $checkpoint();
183                    if ($bytes !== null && !$begun) {
184                        // The first byte is in: the wait for the server is over, and the answer has to arrive within the
185                        // stream timeout from here, whatever its further bytes do.
186                        $begun = true;
187                        $next('Reading the response', 'stream timeout', $streamTimeout);
188                    }
189
190                    return $bytes;
191                });
192                // The response is complete. A timer cannot say that its window has passed while the bytes were decoded, or
193                // while the caller cancelled: the clock and the caller are looked at before it is returned.
194                $checkpoint();
195
196                return $response;
197            },
198        );
199    }
200
201    /**
202     * Runs one IO call in a window of its own, so the time a scan takes after `100 Continue` does not eat
203     * into the time the preview phase had. A call that does not finish leaves a request or a response in
204     * part, and the session is closed for good; whatever ended it is named (the caller's cancellation as the
205     * `Amp\CancelledException` it is, only the window passing as an {@see IcapTimeoutException}). Timer and
206     * subscription are removed on every way out, not left to destructors: an exception that is still
207     * referenced keeps the arguments of its stack frames alive.
208     *
209     * The phase gets the cancellation of the window and a checkpoint for the places where it has control
210     * back from something that does not look by itself: a producer, or a read that returned without the
211     * timer having had a turn of the event loop (it found its bytes in the socket, or something held the
212     * loop up). The checkpoint throws the caller's cancellation, or that of the window once it has passed:
213     * the monotonic clock says so when the timer cannot.
214     *
215     * The phase can go on into a second stretch with a limit of its own: the third argument it is handed
216     * replaces the timer and the end on the clock ($what and $limit are for the message), and the window
217     * stays the one the call and its cancellation belong to. The checkpoint looks at the stretch that runs,
218     * so the first bytes of an answer are measured against the time to begin before the window of the
219     * transfer takes its place.
220     *
221     * @template T
222     *
223     * @param string                                                                         $what    what the call does, for the message
224     * @param string                                                                         $limit   the name of the limit that applies, for the message
225     * @param \Closure(Cancellation, \Closure(): void, \Closure(string, string, float): void): T $phase
226     *
227     * @return T
228     */
229    private function inWindow(string $what, string $limit, float $seconds, \Closure $phase): mixed
230    {
231        $caller = $this->userCancellation;
232        $window = new DeferredCancellation();
233        $expired = false;
234        $deadline = null;
235        $watch = null;
236        // The end of the stretch that runs, on the monotonic clock: $arm sets it together with the timer.
237        $endsAt = 0.0;
238        $checkpoint = static function () use ($caller, $window, &$endsAt, &$expired): void {
239            $caller?->throwIfRequested();
240            if (hrtime(true) >= $endsAt) {
241                $expired = true;
242                $window->cancel();
243            }
244            $window->getCancellation()->throwIfRequested();
245        };
246        $arm = static function (string $newWhat, string $newLimit, float $newSeconds) use (&$what, &$limit, &$seconds, &$endsAt, &$deadline, &$expired, $window): void {
247            if ($deadline !== null) {
248                EventLoop::cancel($deadline);
249            }
250            $what = $newWhat;
251            $limit = $newLimit;
252            $seconds = $newSeconds;
253            // The end on the clock moves with the timer: the checkpoint looks at the stretch that is running.
254            $endsAt = hrtime(true) + $newSeconds * 1e9;
255            $deadline = EventLoop::delay($newSeconds, static function () use (&$expired, $window): void {
256                $expired = true;
257                $window->cancel();
258            });
259        };
260
261        try {
262            // Cancelled before it began: nothing is pulled, sent or read.
263            $caller?->throwIfRequested();
264
265            $arm($what, $limit, $seconds);
266            $watch = $caller?->subscribe(static function () use ($window): void {
267                $window->cancel();
268            });
269
270            return $phase($window->getCancellation(), $checkpoint, $arm);
271        } catch (\Throwable $e) {
272            $this->close();
273
274            // A cancelled window is a cancellation: say whether the caller or the end of the window cancelled it.
275            if ($e instanceof CancelledException) {
276                $caller?->throwIfRequested();
277                if ($expired) {
278                    throw new IcapTimeoutException(sprintf(
279                        '%s did not finish within the %s of %s seconds.',
280                        $what,
281                        $limit,
282                        $seconds,
283                    ));
284                }
285            }
286
287            throw $e;
288        } finally {
289            if ($deadline !== null) {
290                EventLoop::cancel($deadline);
291            }
292            if ($caller !== null && $watch !== null) {
293                $caller->unsubscribe($watch);
294            }
295        }
296    }
297
298    /**
299     * A stream error of the socket is a connection failure, unless the caller or the end of the window ended the call
300     * first (the window closes the socket under a pending write): the checkpoint throws that cancellation.
301     */
302    private static function socketFailed(StreamException $e, \Closure $checkpoint): IcapConnectionException
303    {
304        $checkpoint();
305
306        return new IcapConnectionException('The connection to the server failed.', 0, $e);
307    }
308
309    #[\Override]
310    public function release(): void
311    {
312        if ($this->disposed) {
313            return;
314        }
315        $this->disposed = true;
316
317        // An operation that was cancelled, or has run out of time, hands nothing to the pool, whatever the exchange
318        // looked like while it lasted (M10).
319        if ($this->pool !== null && !($this->userCancellation?->isRequested() ?? false)) {
320            $this->pool->release($this->config, $this->socket);
321            return;
322        }
323        $this->socket->close();
324    }
325
326    #[\Override]
327    public function close(): void
328    {
329        if ($this->disposed) {
330            return;
331        }
332        $this->disposed = true;
333        $this->socket->close();
334    }
335
336    private function assertActive(): void
337    {
338        if ($this->disposed) {
339            throw new \LogicException('TransportSession has already been released or closed.');
340        }
341    }
342}