Code Coverage
 
Lines
Functions and Methods
Classes and Traits
Total
71.23%
52 / 73
62.50%
5 / 8
 CRAP
0.00%
0 / 1
AmpConnectionPool
71.23%
52 / 73
62.50%
5 / 8
 44.35
0.00%
0 / 1
 __construct
78.57%
11 / 14
0.00%
0 / 1
 5.25
 acquire
85.71%
12 / 14
0.00%
0 / 1
 5.07
 release
100.00%
13 / 13
100.00%
1 / 1
 4
 close
100.00%
5 / 5
100.00%
1 / 1
 3
 tuneFromOptions
100.00%
3 / 3
100.00%
1 / 1
 2
 effectiveMaxConnections
100.00%
3 / 3
100.00%
1 / 1
 2
 key
100.00%
5 / 5
100.00%
1 / 1
 2
 defaultConnector
0.00%
0 / 16
0.00%
0 / 1
 20
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\Cancellation;
24use Amp\Socket;
25use Amp\Socket\ConnectContext;
26use Amp\Socket\Socket as SocketInterface;
27use Closure;
28use Ndrstmr\Icap\Config;
29use Ndrstmr\Icap\DTO\IcapResponse;
30use Ndrstmr\Icap\Exception\IcapConnectionException;
31
32/**
33 * In-process keep-alive pool of amphp sockets.
34 *
35 * Idle sockets are stored as a LIFO stack per host:port[:tls] key
36 * (most recently used first — warmer connections are likelier to
37 * still be alive). On {@see acquire()} the pool pops sockets off the
38 * stack, drops any that are already closed, and connects a fresh one
39 * if none are usable. On {@see release()} a socket is pushed back
40 * unless the per-host cap is reached, in which case it's closed.
41 *
42 * Each entry records when it became idle. On {@see acquire()} entries
43 * older than `$maxIdleSeconds` (default 30 s) are evicted and closed
44 * before testing `isClosed()`. This prevents stale socket
45 * accumulation in long-running PHP workers (Swoole, RoadRunner,
46 * ReactPHP) where the server may have closed its end silently.
47 */
48final class AmpConnectionPool implements ConnectionPoolInterface
49{
50    /** @var array<string, list<array{socket: SocketInterface, idleSince: float}>> */
51    private array $idle = [];
52
53    private bool $closed = false;
54
55    /** @var Closure(Config, ?Cancellation): SocketInterface */
56    private Closure $connector;
57
58    /** @var Closure(): float */
59    private Closure $clock;
60
61    /**
62     * @param int                                                                  $maxConnectionsPerHost  cap on idle sockets per host:port[:tls] key
63     * @param (Closure(Config, ?Cancellation): SocketInterface)|null               $connector              optional override — production code uses the amphp connector; tests can inject pre-built socket pairs
64     * @param int|null                                                             $serverMaxConnections   optional server-advertised Max-Connections (RFC 3507 §4.10.2); when set, the effective idle cap becomes min(localCap, serverMax)
65     * @param float                                                                $maxIdleSeconds         idle sockets older than this are evicted on acquire() (default 30 s)
66     * @param (Closure(): float)|null                                              $clock                  monotonic clock for idle-age checks; defaults to microtime(true); injectable for deterministic tests
67     */
68    public function __construct(
69        private int $maxConnectionsPerHost = 8,
70        ?Closure $connector = null,
71        private ?int $serverMaxConnections = null,
72        private float $maxIdleSeconds = 30.0,
73        ?Closure $clock = null,
74    ) {
75        if ($maxConnectionsPerHost < 1) {
76            throw new \InvalidArgumentException(
77                'maxConnectionsPerHost must be >= 1, got: ' . $maxConnectionsPerHost,
78            );
79        }
80
81        if ($serverMaxConnections !== null && $serverMaxConnections < 1) {
82            throw new \InvalidArgumentException(
83                'serverMaxConnections must be >= 1, got: ' . $serverMaxConnections,
84            );
85        }
86
87        if ($maxIdleSeconds <= 0.0) {
88            throw new \InvalidArgumentException(
89                'maxIdleSeconds must be > 0, got: ' . $maxIdleSeconds,
90            );
91        }
92
93        $this->connector = $connector ?? self::defaultConnector();
94        $this->clock = $clock ?? static fn (): float => microtime(true);
95    }
96
97    #[\Override]
98    public function acquire(Config $config, ?Cancellation $cancellation = null): SocketInterface
99    {
100        if ($this->closed) {
101            throw new IcapConnectionException('Pool is closed.');
102        }
103
104        $key = $this->key($config);
105        $now = ($this->clock)();
106
107        while (!empty($this->idle[$key])) {
108            $entry = array_pop($this->idle[$key]);
109            $socket = $entry['socket'];
110
111            if ($socket->isClosed()) {
112                continue;
113            }
114
115            if (($now - $entry['idleSince']) > $this->maxIdleSeconds) {
116                $socket->close();
117                continue;
118            }
119
120            return $socket;
121        }
122
123        return ($this->connector)($config, $cancellation);
124    }
125
126    #[\Override]
127    public function release(Config $config, SocketInterface $socket): void
128    {
129        if ($socket->isClosed() || $this->closed) {
130            $socket->close();
131            return;
132        }
133
134        $key = $this->key($config);
135        $idleCount = count($this->idle[$key] ?? []);
136        $effectiveCap = $this->effectiveMaxConnections();
137
138        if ($idleCount >= $effectiveCap) {
139            $socket->close();
140            return;
141        }
142
143        $this->idle[$key][] = [
144            'socket' => $socket,
145            'idleSince' => ($this->clock)(),
146        ];
147    }
148
149    #[\Override]
150    public function close(): void
151    {
152        $this->closed = true;
153        foreach ($this->idle as $entries) {
154            foreach ($entries as $entry) {
155                $entry['socket']->close();
156            }
157        }
158        $this->idle = [];
159    }
160
161    /**
162     * Extract the `Max-Connections` header from an OPTIONS response and
163     * apply it as the server-side idle cap. Subsequent {@see release()}
164     * calls use `min(localCap, serverMaxConnections)`.
165     *
166     * Typical usage after an OPTIONS round-trip:
167     *
168     *     $result = $client->options('/avscan')->await();
169     *     $pool->tuneFromOptions($result->originalResponse);
170     *
171     * @see \Ndrstmr\Icap\DTO\ScanResult::$originalResponse
172     */
173    public function tuneFromOptions(IcapResponse $response): void
174    {
175        $maxConn = (int) ($response->headers['Max-Connections'][0] ?? '0');
176        if ($maxConn >= 1) {
177            $this->serverMaxConnections = $maxConn;
178        }
179    }
180
181    /**
182     * Effective idle cap: min(localCap, serverMaxConnections) when the
183     * server advertised a Max-Connections header, otherwise localCap.
184     */
185    private function effectiveMaxConnections(): int
186    {
187        if ($this->serverMaxConnections !== null) {
188            return min($this->maxConnectionsPerHost, $this->serverMaxConnections);
189        }
190
191        return $this->maxConnectionsPerHost;
192    }
193
194    private function key(Config $config): string
195    {
196        $key = $config->host . ':' . $config->port;
197        $tls = $config->getTlsContext();
198        if ($tls !== null) {
199            // spl_object_hash() is a deliberate transitional choice: it
200            // guarantees that two *different* ClientTlsContext instances
201            // always produce different pool keys, preventing cross-tenant
202            // socket reuse (CVE-class finding, v2.1.1-A).  The caller is
203            // responsible for not sharing TlsContext objects across tenants.
204            // v2.2 will switch to a deterministic hash derived from the
205            // context's peer name, cert path, and CA bundle so that
206            // equivalent contexts can still share idle connections.
207            $key .= ':tls:' . spl_object_hash($tls);
208        }
209
210        return $key;
211    }
212
213    /**
214     * @return Closure(Config, ?Cancellation): SocketInterface
215     */
216    private static function defaultConnector(): Closure
217    {
218        return static function (Config $config, ?Cancellation $cancellation): SocketInterface {
219            $tls = $config->getTlsContext();
220            $url = sprintf('tcp://%s:%d', $config->host, $config->port);
221            $context = (new ConnectContext())->withConnectTimeout($config->getSocketTimeout());
222            if ($tls !== null) {
223                $context = $context->withTlsContext($tls);
224            }
225
226            try {
227                if ($tls !== null) {
228                    return Socket\connectTls($url, $context, $cancellation);
229                }
230                return Socket\connect($url, $context, $cancellation);
231            } catch (Socket\ConnectException $e) {
232                throw new IcapConnectionException(
233                    sprintf('Pooled connect to %s:%d failed.', $config->host, $config->port),
234                    0,
235                    $e,
236                );
237            }
238        };
239    }
240}