From fc54711d09f0ff82757332ad1d021a38212696b7 Mon Sep 17 00:00:00 2001 From: loks0n <22452787+loks0n@users.noreply.github.com> Date: Mon, 27 Jul 2026 17:14:32 +0100 Subject: [PATCH] feat(pools)!: constructor-only configuration, one acquisition budget MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Five interacting knobs decided how long pop() could block, and nothing named the total. reconnectAttempts x reconnectSleep inside retryAttempts x retrySleep, plus synchronizedTimeout on each round, at the shipped defaults meant a pool fronting an unreachable backend took around 44 seconds to fail a single acquisition against a 3 second connect timeout. Nobody chose 44 seconds; it was the product of defaults nobody had multiplied out. Consumers each picked their own numbers — one sets reconnectAttempts to 15, giving roughly three minutes — and the pattern drifts because a default plus an optional override always drifts. Replace all five with `timeout`: one required constructor argument that is the whole budget a caller may spend. pop() creates when there is capacity, otherwise waits out the remaining budget on the adapter, then throws. One number, and it is the number a caller experiences. It is the pool's only timeout, and it matches the name Adapter::pop() already used for the same concept. Creation retries are gone with no replacement. Retrying a failed connect belongs to the init callback, because only the caller knows whether a failure is worth retrying and on what schedule, and a circuit breaker composed there carries state between calls in a way an attempt counter cannot. Retrying inside the pool was also actively harmful: the capacity slot is reserved for the whole attempt sequence, so retries turned a shared resource into a private speculative wait. Configuration is now constructor-only readonly state, per the package-design skill. Twelve setters are removed across Pool, Connection and Group, along with the getters that only existed to read them back. Pool drops from 23 public methods to 10, Connection from 9 to 3. Connection is final readonly. Identity, resource and owning pool are fixed at construction, so a connection cannot be repointed while another coroutine holds it, and because the pool is no longer nullable, returning one cannot fail for want of an owner — two exception paths disappear. Other changes that fell out of the review: - destroy() releases capacity only for a connection this pool is actually holding. It decremented unconditionally, so destroying the same connection twice, or one belonging to another pool, drove the count below the truth and let the pool create beyond its size. - A failed create no longer falls through to waiting out the remaining budget. Waiting for another caller's connection after your own create failed is a retry in disguise, which this change exists to remove, and it contradicted the documented behaviour. init's exception now propagates untouched, so callers keep the type they need to act on. - timeout is documented as bounding the wait for an available connection, not the total call. The pool cannot interrupt a blocking connect, so init owns its own connect timeout; claiming otherwise promised something unenforceable. - destroy() no longer creates the replacement itself. It frees capacity and the next pop() creates one, which matches the pool's documented lazy initialisation and stops destroy() blocking on a connect or failing for reasons unrelated to the connection being discarded. - The pool measures use time itself, keyed by connection id, so release() no longer takes a $start timestamp from its caller. Group gets correct per-pool timing without threading timestamps through. - The Swoole adapter clamped a non-positive channel timeout, which Swoole reads as "wait forever" — the exact opposite of a zero budget, and a hang the old retry loop hid. Its docblock claimed 0 meant non-blocking. - Adapter::pop() takes a float timeout, so sub-second budgets survive the seam. The contract's docblocks now say what a non-concurrent adapter honestly does instead of apologising for ignoring arguments. - init is typed Closure rather than callable, so TResource survives into Connection without an inline annotation. - Guards for size and timeout, throwing rather than annotating. - phpstan at level max, clean without a single ignore. Rector extends the root with the remaining stable prepared sets. - README documents the acquisition budget and where retries belong, with an "Upgrading from 1.x" section mapping every removed API. The Swoole suite's six snake_case test methods are renamed to camelCase, matching every other test in the package. Dependents are updated in the same commit. The client and queue test suites construct pools directly, and client's factory helper takes a Closure to match the tightened init type. Messaging built its pool with the adapter passed as `pool:` and sized it `min(count($requests), MAX)`, which is zero for an empty batch and now trips the size guard, so it passes `adapter:`, clamps the size and states its timeout. All three constrain `^2.0` and pass --linked. Per the note in the tests workflow, that leaves the registry combination unbuildable until pools/2.0.0 is tagged, which is why dependents run linked here. Net 386 lines lighter in pools. Its suite runs in 0.4s rather than 47s, because the sleeps it used to spend were the point. BREAKING CHANGE: see "Upgrading from 1.x" in packages/pools/README.md. Co-Authored-By: Claude Opus 5 (1M context) --- .../config/vocabularies/Utopia/accept.txt | 10 +- packages/client/composer.json | 2 +- packages/client/tests/Client/PoolTest.php | 6 +- packages/messaging/composer.json | 2 +- .../src/Utopia/Messaging/Adapter.php | 7 +- packages/pools/README.md | 150 +++-- packages/pools/composer.lock | 144 ++--- packages/pools/phpstan.neon | 2 +- packages/pools/rector.php | 16 +- packages/pools/src/Pools/Adapter.php | 31 +- packages/pools/src/Pools/Adapter/Stack.php | 14 +- packages/pools/src/Pools/Adapter/Swoole.php | 25 +- packages/pools/src/Pools/Connection.php | 105 +--- packages/pools/src/Pools/Group.php | 74 +-- packages/pools/src/Pools/Pool.php | 564 +++++++----------- .../pools/tests/Pools/Adapter/StackTest.php | 4 +- .../pools/tests/Pools/Adapter/SwooleTest.php | 142 +++-- packages/pools/tests/Pools/Base.php | 4 +- .../Pools/Scopes/ConnectionTestScope.php | 119 +--- .../tests/Pools/Scopes/GroupTestScope.php | 74 +-- .../tests/Pools/Scopes/PoolTestScope.php | 346 +++++------ packages/queue/composer.json | 2 +- .../tests/Queue/E2E/Adapter/PoolTest.php | 2 +- 23 files changed, 759 insertions(+), 1086 deletions(-) diff --git a/.vale/styles/config/vocabularies/Utopia/accept.txt b/.vale/styles/config/vocabularies/Utopia/accept.txt index 84f36abaf..cb1cbbbcf 100644 --- a/.vale/styles/config/vocabularies/Utopia/accept.txt +++ b/.vale/styles/config/vocabularies/Utopia/accept.txt @@ -1,5 +1,6 @@ - - +# Brands, products, people +# Protocols, formats, acronyms +# Tech vocabulary [Cc]harset(?:s)? [Cc]onfig(?:s)? [Cc]ron @@ -13,9 +14,6 @@ [Ss]tdout [Ss]ubtree [Vv]alidator(?:s)? -# Brands, products, people -# Protocols, formats, acronyms -# Tech vocabulary accessor(?:s)? ACLs? aiohttp @@ -68,6 +66,7 @@ Fedora FPM Fux GET +getters? GPTBot Grafana Guzzle @@ -148,6 +147,7 @@ Sailfish sanitization Scrypt Sentry +setters? Silk Sinch SMS diff --git a/packages/client/composer.json b/packages/client/composer.json index d0b49e157..df897efb5 100644 --- a/packages/client/composer.json +++ b/packages/client/composer.json @@ -30,7 +30,7 @@ "psr/http-client": "^1.0", "psr/http-factory": "^1.0", "psr/http-message": "^1.1 || ^2.0", - "utopia-php/pools": "^1.0", + "utopia-php/pools": "^2.0", "utopia-php/psr7": "^0.2", "utopia-php/span": "^1.1 || ^3.0 || ^4.0" }, diff --git a/packages/client/tests/Client/PoolTest.php b/packages/client/tests/Client/PoolTest.php index 2e5a81b0d..5154368f4 100644 --- a/packages/client/tests/Client/PoolTest.php +++ b/packages/client/tests/Client/PoolTest.php @@ -58,13 +58,13 @@ public function testItReclaimsTheConnectionSoItCanBeReused(): void * * @template T of \Psr\Http\Client\ClientInterface&StreamingClientInterface * - * @param callable(): T $init + * @param \Closure(): T $init * * @return Connections */ - private function connections(callable $init, int $size = 4): Connections + private function connections(\Closure $init, int $size = 4): Connections { - return new Connections(new Stack(), 'test', $size, $init); + return new Connections(new Stack(), 'test', $size, $init, timeout: 0.0); } private function request(): RequestInterface diff --git a/packages/messaging/composer.json b/packages/messaging/composer.json index 6a3c393f3..c25767159 100644 --- a/packages/messaging/composer.json +++ b/packages/messaging/composer.json @@ -36,7 +36,7 @@ "psr/http-client": "^1.0", "psr/http-message": "^1.1 || ^2.0", "utopia-php/client": "^0.2", - "utopia-php/pools": "^1.0", + "utopia-php/pools": "^2.0", "utopia-php/telemetry": "^0.4" }, "require-dev": { diff --git a/packages/messaging/src/Utopia/Messaging/Adapter.php b/packages/messaging/src/Utopia/Messaging/Adapter.php index 4c12715e4..229cbbfdb 100644 --- a/packages/messaging/src/Utopia/Messaging/Adapter.php +++ b/packages/messaging/src/Utopia/Messaging/Adapter.php @@ -271,10 +271,13 @@ protected function requestMulti( $run = function () use ($requests, $timeout, $connectTimeout, &$results): void { $pool = new ConnectionPool( - pool: new SwoolePoolAdapter(), + adapter: new SwoolePoolAdapter(), name: self::CONNECTION_POOL_NAME, - size: min(\count($requests), self::MAX_CONCURRENT_REQUESTS), + size: max(1, min(\count($requests), self::MAX_CONCURRENT_REQUESTS)), init: $this->clientFactory ?? $this->defaultClient($timeout, $connectTimeout)->withConnectionReuse(...), + // A slot per request, so acquisition never queues; the request + // timeouts belong to the client, not to getting hold of one. + timeout: 0.0, ); $group = new WaitGroup(); diff --git a/packages/pools/README.md b/packages/pools/README.md index 637414be7..41fea79a3 100644 --- a/packages/pools/README.md +++ b/packages/pools/README.md @@ -27,79 +27,117 @@ composer require utopia-php/pools ```php use PDO; -use Utopia\Pools\Pool; +use Utopia\Pools\Adapter\Swoole; use Utopia\Pools\Group; +use Utopia\Pools\Pool; -$pool = new Pool('mysql-pool', 1 /* number of connections */, function() { - $host = '127.0.0.1'; - $db = 'test'; - $user = 'root'; - $pass = ''; - $charset = 'utf8mb4'; - - try { - $pdo = new PDO("mysql:host=$host;dbname=$db;charset=$charset", $user, $pass); - } catch (\PDOException $e) { - throw new \PDOException($e->getMessage(), (int)$e->getCode()); - } - - return $pdo; -}); - -$pool->setReconnectAttempts(3); // number of attempts to reconnect -$pool->setReconnectSleep(5); // seconds to sleep between reconnect attempts - -$pool->setRetryAttempts(3); // number of attempts to get connection -$pool->setRetrySleep(5); // seconds to sleep between failed pop-connection attempts - -$connection = $pool->pop(); // Get a connection from the pool -$connection->getID(); // Get the connection ID -$connection->getResource(); // Get the connection resource - -$pool->push($connection); // Return the connection to the pool +$pool = new Pool( + adapter: new Swoole, + name: 'mysql-pool', + size: 8, + init: fn () => new PDO('mysql:host=127.0.0.1;dbname=test;charset=utf8mb4', 'root', ''), + timeout: 2.0, +); + +// Preferred: the pool hands the resource to the callback and takes it back +// afterwards, discarding it if the callback threw. +$rows = $pool->use(fn (PDO $pdo) => $pdo->query('SELECT 1')->fetchAll()); + +// Manual lifetime, when `use()` does not fit. +$connection = $pool->pop(); +$connection->id; // "mysql-pool-6885a1f2c4e19" +$connection->resource; // the PDO instance +$connection->reclaim(); // return it for reuse +$connection->destroy(); // or discard it and free the capacity + +$pool->count(); // connections a caller could still obtain +$pool->isEmpty(); // none available without one being returned first +$pool->isFull(); // nothing checked out + +$group = new Group; +$group->add($pool); +$group->get('mysql-pool'); +$group->use(['mysql-pool'], fn (PDO $pdo) => $pdo->query('SELECT 1')); +``` -$pool->reclaim(); // Recalim the pool, return all active connections automatically +## Acquiring connections -$pool->count(); // Get the number of available connections +`timeout` bounds how long `pop()` waits for a connection to become available. +The pool creates one when it has spare capacity, otherwise it waits for one to +come back, and it throws once the budget is spent. There is one number for +waiting, and it is the number a caller experiences. -$pool->isEmpty(); // Check if the pool is empty +It does not cover time spent inside `init`. The pool cannot interrupt a blocking +connect, so give `init` its own connect timeout if you need the total bounded — +for PDO that is `PDO::ATTR_TIMEOUT`. -$pool->isFull(); // Check if the pool is full +If creating a connection fails, that exception propagates untouched, so callers +keep the original type to act on. The pool does not retry and does not fall back +to waiting, because waiting for another caller's connection after your own +create failed is a retry in disguise. Own retries in `init`: -$group = new Group(); // Create a group of pools -$group->add($pool); // Add a pool to the group -$group->get('mysql-pool'); // Get a pool from the group -$group->setReconnectAttempts(3); // Set the number of reconnect attempts for all pools -$group->setReconnectSleep(5); // Set the sleep time between reconnect attempts for all pools +```php +$pool = new Pool( + adapter: new Swoole, + name: 'mysql-pool', + size: 8, + init: fn () => $breaker->call(fn () => new PDO(/* ... */)), + timeout: 2.0, +); ``` -## Reconnect and retry - -Both reconnect and retry logic is used in `pop()` method to handle problematic scenarios. Both allow you to configure 2 properties: - -- `attempts` - How many times library will retry when problem occurs -- `sleep` - How long will library wait for before next retry attempt - -**Reconnect** settings are used when your connection initialization callback throws an exception. This can occur for example, when a handshake with SQL server fails. - -**Retry** settings are used when pool of connection is empty and there is no more connections to pop. This can occur for example, when your server supports more concurrent actions than your pool. +Compose [`utopia-php/circuit-breaker`](https://github.com/utopia-php/circuit-breaker) +there rather than counting attempts, because a breaker carries state between +calls and an attempt counter cannot. + +## Upgrading from 1.x + +Configuration is constructor-only, and the retry knobs are gone. + +- `timeout` is a required constructor argument, and it is the pool's only + timeout. It replaces `setRetryAttempts()`, `setRetrySleep()` and + `setSynchronizationTimeout()`. Pass the total wait you want, in seconds. +- `setReconnectAttempts()` and `setReconnectSleep()` are removed with no + replacement. Wrap `init` if you want to retry a failed connect. +- `setTelemetry()` is removed. Pass `telemetry:` to the constructor. +- `Group::setReconnectAttempts()`, `Group::setReconnectSleep()` and + `Group::setTelemetry()` are removed. Pools arrive configured. +- `Pool::getName()` and `Pool::getSize()` are now the `$name` and `$size` + properties. The `getReconnectAttempts()`, `getReconnectSleep()`, + `getRetryAttempts()`, `getRetrySleep()` and `getSynchronizationTimeout()` + getters are removed. +- `Connection` is `final readonly`. `getID()`, `getResource()` and `getPool()` + become the `$id` and `$resource` properties; `getPool()` has no replacement. + `setID()`, `setResource()` and `setPool()` are removed. +- `Connection::reclaim()` and `Connection::destroy()` return `void` instead of + the pool. +- `Pool::release()` no longer takes a `$start` timestamp. The pool measures use + time itself. +- `Adapter::pop()` takes a `float` timeout. Custom adapters need the wider type. +- `init` is typed `Closure` rather than `callable`. Pass a closure, or wrap a + string or array callable with `Closure::fromCallable()`. + +Two behaviour changes worth checking against your own code: + +- A failed connect used to be absorbed by an internal retry and can now reach + the caller on the first attempt, as whatever type `init` threw rather than a + pool exception. +- `destroy()` no longer creates the replacement connection itself. Capacity is + freed immediately and the next `pop()` creates one, so `destroy()` cannot + block on a connect or fail for reasons unrelated to the connection you asked + it to discard. ## System requirements -Utopia Framework requires PHP 8.0 or later. We recommend using the latest PHP version whenever possible. +Utopia pools requires PHP 8.4 or later. We recommend using the latest PHP version whenever possible. ## Tests -To run all unit tests, use the following Docker command: - -```bash -docker compose exec tests vendor/bin/phpunit --configuration phpunit.xml tests -``` - -To run static code analysis, use the following Psalm command: +Run the test suite and static analysis from the monorepo root: ```bash -docker compose exec tests vendor/bin/psalm --show-info=true +bin/monorepo test pools +bin/monorepo check pools ``` ## Copyright and license diff --git a/packages/pools/composer.lock b/packages/pools/composer.lock index c3a6d034a..1af063813 100644 --- a/packages/pools/composer.lock +++ b/packages/pools/composer.lock @@ -8,23 +8,22 @@ "packages": [ { "name": "brick/math", - "version": "0.14.8", + "version": "0.18.0", "source": { "type": "git", "url": "https://github.com/brick/math.git", - "reference": "63422359a44b7f06cae63c3b429b59e8efcc0629" + "reference": "82944324d1c1bdb2c2618e89978d4e2ad78d69ad" }, "dist": { "type": "zip", - "url": "https://api.github.com/repos/brick/math/zipball/63422359a44b7f06cae63c3b429b59e8efcc0629", - "reference": "63422359a44b7f06cae63c3b429b59e8efcc0629", + "url": "https://api.github.com/repos/brick/math/zipball/82944324d1c1bdb2c2618e89978d4e2ad78d69ad", + "reference": "82944324d1c1bdb2c2618e89978d4e2ad78d69ad", "shasum": "" }, "require": { "php": "^8.2" }, "require-dev": { - "php-coveralls/php-coveralls": "^2.2", "phpstan/phpstan": "2.1.22", "phpunit/phpunit": "^11.5" }, @@ -56,7 +55,7 @@ ], "support": { "issues": "https://github.com/brick/math/issues", - "source": "https://github.com/brick/math/tree/0.14.8" + "source": "https://github.com/brick/math/tree/0.18.0" }, "funding": [ { @@ -64,7 +63,7 @@ "type": "github" } ], - "time": "2026-02-10T14:33:43+00:00" + "time": "2026-06-14T18:21:03+00:00" }, { "name": "composer/semver", @@ -145,23 +144,23 @@ }, { "name": "google/protobuf", - "version": "v4.33.6", + "version": "v5.35.1", "source": { "type": "git", "url": "https://github.com/protocolbuffers/protobuf-php.git", - "reference": "84b008c23915ed94536737eae46f41ba3bccfe67" + "reference": "55bb4a7d6739b5af0927b96213c1371a3afb7cfb" }, "dist": { "type": "zip", - "url": "https://api.github.com/repos/protocolbuffers/protobuf-php/zipball/84b008c23915ed94536737eae46f41ba3bccfe67", - "reference": "84b008c23915ed94536737eae46f41ba3bccfe67", + "url": "https://api.github.com/repos/protocolbuffers/protobuf-php/zipball/55bb4a7d6739b5af0927b96213c1371a3afb7cfb", + "reference": "55bb4a7d6739b5af0927b96213c1371a3afb7cfb", "shasum": "" }, "require": { - "php": ">=8.1.0" + "php": ">=8.2.0" }, "require-dev": { - "phpunit/phpunit": ">=10.5.62 <11.0.0" + "phpunit/phpunit": ">=11.5.0 <12.0.0" }, "suggest": { "ext-bcmath": "Need to support JSON deserialization" @@ -183,9 +182,9 @@ "proto" ], "support": { - "source": "https://github.com/protocolbuffers/protobuf-php/tree/v4.33.6" + "source": "https://github.com/protocolbuffers/protobuf-php/tree/v5.35.1" }, - "time": "2026-03-18T17:32:05+00:00" + "time": "2026-06-11T21:19:23+00:00" }, { "name": "nyholm/psr7", @@ -333,16 +332,16 @@ }, { "name": "open-telemetry/api", - "version": "1.9.0", + "version": "1.10.0", "source": { "type": "git", "url": "https://github.com/opentelemetry-php/api.git", - "reference": "6f8d237ce2c304ca85f31970f788e7f074d147be" + "reference": "7c029c4a6fd457094a20569bf98f93d95e9a7559" }, "dist": { "type": "zip", - "url": "https://api.github.com/repos/opentelemetry-php/api/zipball/6f8d237ce2c304ca85f31970f788e7f074d147be", - "reference": "6f8d237ce2c304ca85f31970f788e7f074d147be", + "url": "https://api.github.com/repos/opentelemetry-php/api/zipball/7c029c4a6fd457094a20569bf98f93d95e9a7559", + "reference": "7c029c4a6fd457094a20569bf98f93d95e9a7559", "shasum": "" }, "require": { @@ -399,7 +398,7 @@ "issues": "https://github.com/open-telemetry/opentelemetry-php/issues", "source": "https://github.com/open-telemetry/opentelemetry-php" }, - "time": "2026-02-25T13:24:05+00:00" + "time": "2026-07-06T12:28:04+00:00" }, { "name": "open-telemetry/context", @@ -526,20 +525,20 @@ }, { "name": "open-telemetry/gen-otlp-protobuf", - "version": "1.9.0", + "version": "1.10.0", "source": { "type": "git", "url": "https://github.com/opentelemetry-php/gen-otlp-protobuf.git", - "reference": "a229cf161d42001d64c8f21e8f678581fe1c66b9" + "reference": "66f04d0e448ad333033bfc7baae1aa56330be088" }, "dist": { "type": "zip", - "url": "https://api.github.com/repos/opentelemetry-php/gen-otlp-protobuf/zipball/a229cf161d42001d64c8f21e8f678581fe1c66b9", - "reference": "a229cf161d42001d64c8f21e8f678581fe1c66b9", + "url": "https://api.github.com/repos/opentelemetry-php/gen-otlp-protobuf/zipball/66f04d0e448ad333033bfc7baae1aa56330be088", + "reference": "66f04d0e448ad333033bfc7baae1aa56330be088", "shasum": "" }, "require": { - "google/protobuf": "^3.22 || ^4.0", + "google/protobuf": "^3.22 || ^4.0 || ^5.0", "php": "^8.0" }, "suggest": { @@ -585,20 +584,20 @@ "issues": "https://github.com/open-telemetry/opentelemetry-php/issues", "source": "https://github.com/open-telemetry/opentelemetry-php" }, - "time": "2025-10-19T06:44:33+00:00" + "time": "2026-06-17T12:06:32+00:00" }, { "name": "open-telemetry/sdk", - "version": "1.14.0", + "version": "1.15.0", "source": { "type": "git", "url": "https://github.com/opentelemetry-php/sdk.git", - "reference": "6e3d0ce93e76555dd5e2f1d19443ff45b990e410" + "reference": "77e1aa73850154abb86937d52a70883edc3b4547" }, "dist": { "type": "zip", - "url": "https://api.github.com/repos/opentelemetry-php/sdk/zipball/6e3d0ce93e76555dd5e2f1d19443ff45b990e410", - "reference": "6e3d0ce93e76555dd5e2f1d19443ff45b990e410", + "url": "https://api.github.com/repos/opentelemetry-php/sdk/zipball/77e1aa73850154abb86937d52a70883edc3b4547", + "reference": "77e1aa73850154abb86937d52a70883edc3b4547", "shasum": "" }, "require": { @@ -606,7 +605,7 @@ "nyholm/psr7-server": "^1.1", "open-telemetry/api": "^1.8", "open-telemetry/context": "^1.4", - "open-telemetry/sem-conv": "^1.0", + "open-telemetry/sem-conv": "^1.38.0", "php": "^8.1", "php-http/discovery": "^1.14", "psr/http-client": "^1.0", @@ -629,7 +628,10 @@ "spi": { "OpenTelemetry\\API\\Configuration\\ConfigEnv\\EnvComponentLoader": [ "OpenTelemetry\\API\\Instrumentation\\Configuration\\General\\ConfigEnv\\EnvComponentLoaderHttpConfig", - "OpenTelemetry\\API\\Instrumentation\\Configuration\\General\\ConfigEnv\\EnvComponentLoaderPeerConfig" + "OpenTelemetry\\API\\Instrumentation\\Configuration\\General\\ConfigEnv\\EnvComponentLoaderPeerConfig", + "OpenTelemetry\\SDK\\ConfigEnv\\Trace\\SpanSuppressionStrategySemConv", + "OpenTelemetry\\SDK\\ConfigEnv\\Trace\\SpanSuppressionStrategySpanKind", + "OpenTelemetry\\SDK\\ConfigEnv\\Distribution\\DistributionConfigurationSdk" ], "OpenTelemetry\\SDK\\Common\\Configuration\\Resolver\\ResolverInterface": [ "OpenTelemetry\\SDK\\Common\\Configuration\\Resolver\\SdkConfigurationResolver" @@ -639,7 +641,7 @@ ] }, "branch-alias": { - "dev-main": "1.12.x-dev" + "dev-main": "1.14.x-dev" } }, "autoload": { @@ -682,7 +684,7 @@ "issues": "https://github.com/open-telemetry/opentelemetry-php/issues", "source": "https://github.com/open-telemetry/opentelemetry-php" }, - "time": "2026-03-21T11:50:01+00:00" + "time": "2026-07-14T13:09:54+00:00" }, { "name": "open-telemetry/sem-conv", @@ -1161,20 +1163,20 @@ }, { "name": "ramsey/uuid", - "version": "4.9.2", + "version": "4.9.3", "source": { "type": "git", "url": "https://github.com/ramsey/uuid.git", - "reference": "8429c78ca35a09f27565311b98101e2826affde0" + "reference": "1df15849d00943a67d677dc9cfd80795f038c9f8" }, "dist": { "type": "zip", - "url": "https://api.github.com/repos/ramsey/uuid/zipball/8429c78ca35a09f27565311b98101e2826affde0", - "reference": "8429c78ca35a09f27565311b98101e2826affde0", + "url": "https://api.github.com/repos/ramsey/uuid/zipball/1df15849d00943a67d677dc9cfd80795f038c9f8", + "reference": "1df15849d00943a67d677dc9cfd80795f038c9f8", "shasum": "" }, "require": { - "brick/math": "^0.8.16 || ^0.9 || ^0.10 || ^0.11 || ^0.12 || ^0.13 || ^0.14", + "brick/math": ">=0.8.16 <=0.18", "php": "^8.0", "ramsey/collection": "^1.2 || ^2.0" }, @@ -1233,22 +1235,22 @@ ], "support": { "issues": "https://github.com/ramsey/uuid/issues", - "source": "https://github.com/ramsey/uuid/tree/4.9.2" + "source": "https://github.com/ramsey/uuid/tree/4.9.3" }, - "time": "2025-12-14T04:43:48+00:00" + "time": "2026-06-18T03:57:49+00:00" }, { "name": "symfony/deprecation-contracts", - "version": "v3.7.0", + "version": "v3.7.1", "source": { "type": "git", "url": "https://github.com/symfony/deprecation-contracts.git", - "reference": "50f59d1f3ca46d41ac911f97a78626b6756af35b" + "reference": "f3202fa1b5097b0af062dc978b32ecf63404e31d" }, "dist": { "type": "zip", - "url": "https://api.github.com/repos/symfony/deprecation-contracts/zipball/50f59d1f3ca46d41ac911f97a78626b6756af35b", - "reference": "50f59d1f3ca46d41ac911f97a78626b6756af35b", + "url": "https://api.github.com/repos/symfony/deprecation-contracts/zipball/f3202fa1b5097b0af062dc978b32ecf63404e31d", + "reference": "f3202fa1b5097b0af062dc978b32ecf63404e31d", "shasum": "" }, "require": { @@ -1286,7 +1288,7 @@ "description": "A generic function and convention to trigger deprecation notices", "homepage": "https://symfony.com", "support": { - "source": "https://github.com/symfony/deprecation-contracts/tree/v3.7.0" + "source": "https://github.com/symfony/deprecation-contracts/tree/v3.7.1" }, "funding": [ { @@ -1306,20 +1308,20 @@ "type": "tidelift" } ], - "time": "2026-04-13T15:52:40+00:00" + "time": "2026-06-05T06:23:12+00:00" }, { "name": "symfony/http-client", - "version": "v7.4.13", + "version": "v7.4.14", "source": { "type": "git", "url": "https://github.com/symfony/http-client.git", - "reference": "e8a112b8415707265a7e614278136a9d92989a6a" + "reference": "f6bc6b5a54ff5afac4725cacec9bf2f52eb15920" }, "dist": { "type": "zip", - "url": "https://api.github.com/repos/symfony/http-client/zipball/e8a112b8415707265a7e614278136a9d92989a6a", - "reference": "e8a112b8415707265a7e614278136a9d92989a6a", + "url": "https://api.github.com/repos/symfony/http-client/zipball/f6bc6b5a54ff5afac4725cacec9bf2f52eb15920", + "reference": "f6bc6b5a54ff5afac4725cacec9bf2f52eb15920", "shasum": "" }, "require": { @@ -1387,7 +1389,7 @@ "http" ], "support": { - "source": "https://github.com/symfony/http-client/tree/v7.4.13" + "source": "https://github.com/symfony/http-client/tree/v7.4.14" }, "funding": [ { @@ -1407,20 +1409,20 @@ "type": "tidelift" } ], - "time": "2026-05-24T09:57:54+00:00" + "time": "2026-06-16T11:50:14+00:00" }, { "name": "symfony/http-client-contracts", - "version": "v3.7.0", + "version": "v3.7.1", "source": { "type": "git", "url": "https://github.com/symfony/http-client-contracts.git", - "reference": "4a2d00c37651c0bdc2b9e1c773487a8bf4edb12d" + "reference": "41fc42d276aeff21192465331ebbab7d83a743c0" }, "dist": { "type": "zip", - "url": "https://api.github.com/repos/symfony/http-client-contracts/zipball/4a2d00c37651c0bdc2b9e1c773487a8bf4edb12d", - "reference": "4a2d00c37651c0bdc2b9e1c773487a8bf4edb12d", + "url": "https://api.github.com/repos/symfony/http-client-contracts/zipball/41fc42d276aeff21192465331ebbab7d83a743c0", + "reference": "41fc42d276aeff21192465331ebbab7d83a743c0", "shasum": "" }, "require": { @@ -1469,7 +1471,7 @@ "standards" ], "support": { - "source": "https://github.com/symfony/http-client-contracts/tree/v3.7.0" + "source": "https://github.com/symfony/http-client-contracts/tree/v3.7.1" }, "funding": [ { @@ -1489,7 +1491,7 @@ "type": "tidelift" } ], - "time": "2026-03-06T13:17:50+00:00" + "time": "2026-06-05T06:23:12+00:00" }, { "name": "symfony/polyfill-mbstring", @@ -1738,16 +1740,16 @@ }, { "name": "symfony/service-contracts", - "version": "v3.7.0", + "version": "v3.7.1", "source": { "type": "git", "url": "https://github.com/symfony/service-contracts.git", - "reference": "d25d82433a80eba6aa0e6c24b61d7370d99e444a" + "reference": "c0a284bab1ed8aa0417e3d69250ab437739563a0" }, "dist": { "type": "zip", - "url": "https://api.github.com/repos/symfony/service-contracts/zipball/d25d82433a80eba6aa0e6c24b61d7370d99e444a", - "reference": "d25d82433a80eba6aa0e6c24b61d7370d99e444a", + "url": "https://api.github.com/repos/symfony/service-contracts/zipball/c0a284bab1ed8aa0417e3d69250ab437739563a0", + "reference": "c0a284bab1ed8aa0417e3d69250ab437739563a0", "shasum": "" }, "require": { @@ -1801,7 +1803,7 @@ "standards" ], "support": { - "source": "https://github.com/symfony/service-contracts/tree/v3.7.0" + "source": "https://github.com/symfony/service-contracts/tree/v3.7.1" }, "funding": [ { @@ -1821,7 +1823,7 @@ "type": "tidelift" } ], - "time": "2026-03-28T09:44:51+00:00" + "time": "2026-06-16T09:55:08+00:00" }, { "name": "tbachert/spi", @@ -1930,16 +1932,16 @@ "packages-dev": [ { "name": "swoole/ide-helper", - "version": "6.0.2", + "version": "6.2.2", "source": { "type": "git", "url": "https://github.com/swoole/ide-helper.git", - "reference": "6f12243dce071714c5febe059578d909698f9a52" + "reference": "ad91a281e135f498426fb2e5297376581a4e2dca" }, "dist": { "type": "zip", - "url": "https://api.github.com/repos/swoole/ide-helper/zipball/6f12243dce071714c5febe059578d909698f9a52", - "reference": "6f12243dce071714c5febe059578d909698f9a52", + "url": "https://api.github.com/repos/swoole/ide-helper/zipball/ad91a281e135f498426fb2e5297376581a4e2dca", + "reference": "ad91a281e135f498426fb2e5297376581a4e2dca", "shasum": "" }, "type": "library", @@ -1956,9 +1958,9 @@ "description": "IDE help files for Swoole.", "support": { "issues": "https://github.com/swoole/ide-helper/issues", - "source": "https://github.com/swoole/ide-helper/tree/6.0.2" + "source": "https://github.com/swoole/ide-helper/tree/6.2.2" }, - "time": "2025-03-23T07:31:41+00:00" + "time": "2026-07-25T08:15:29+00:00" } ], "aliases": [], diff --git a/packages/pools/phpstan.neon b/packages/pools/phpstan.neon index 85ce10968..68f401f1b 100644 --- a/packages/pools/phpstan.neon +++ b/packages/pools/phpstan.neon @@ -2,6 +2,6 @@ includes: - ../../phpstan.neon parameters: - level: 8 + level: max scanDirectories: - vendor/swoole/ide-helper diff --git a/packages/pools/rector.php b/packages/pools/rector.php index c56edd2c0..35a483abe 100644 --- a/packages/pools/rector.php +++ b/packages/pools/rector.php @@ -2,12 +2,10 @@ declare(strict_types=1); -use Rector\Config\RectorConfig; - -return RectorConfig::configure() - ->withPaths([ - __DIR__ . '/src', - __DIR__ . '/tests', - ]) - ->withPhpSets() - ->withTypeCoverageLevel(0); +// Everything the root baseline runs, plus the remaining stable prepared sets. +return (require __DIR__ . '/../../rector.php')->withPreparedSets( + typeDeclarationDocblocks: true, + privatization: true, + instanceOf: true, + rectorPreset: true, +); diff --git a/packages/pools/src/Pools/Adapter.php b/packages/pools/src/Pools/Adapter.php index 675d4c3dc..1c8a64fde 100644 --- a/packages/pools/src/Pools/Adapter.php +++ b/packages/pools/src/Pools/Adapter.php @@ -1,26 +1,45 @@ $pool */ + /** @var array */ protected array $pool = []; /** @@ -17,11 +19,12 @@ class Stack extends Adapter * - The stack adapter does NOT enforce capacity limits. * - `$size` is ignored because the pool is backed by a simple array. * - * @param int $size Ignored by the stack adapter. + * @param int $size Ignored by the stack adapter. */ public function initialize(int $size): static { $this->pool = []; + return $this; } @@ -29,6 +32,7 @@ public function push(mixed $connection): static { // Push connection to pool $this->pool[] = $connection; + return $this; } @@ -38,10 +42,10 @@ public function push(mixed $connection): static * Note: The stack adapter does not support blocking operations. * The `$timeout` parameter is ignored. * - * @param int $timeout Ignored by the stack adapter. + * @param float $timeout Ignored by the stack adapter. * @return mixed|null Returns the popped item, or null if the stack is empty. */ - public function pop(int $timeout): mixed + public function pop(float $timeout): mixed { return array_pop($this->pool); } @@ -56,7 +60,7 @@ public function count(): int * * This implementation does not provide mutual exclusion. * - * @param callable $callback Callback to execute. + * @param callable $callback Callback to execute. * @return mixed The value returned by the callback. */ public function synchronized(callable $callback): mixed diff --git a/packages/pools/src/Pools/Adapter/Swoole.php b/packages/pools/src/Pools/Adapter/Swoole.php index 7f55da81b..75cae1c3b 100644 --- a/packages/pools/src/Pools/Adapter/Swoole.php +++ b/packages/pools/src/Pools/Adapter/Swoole.php @@ -1,5 +1,7 @@ pool->push($connection); + return $this; } /** - * Pop an item from the pool. + * Pop an item from the pool, waiting up to $timeout seconds. * - * @param int $timeout Timeout in seconds. Use 0 for non-blocking pop. - * @return mixed|false Returns the pooled value, or false if the pool is empty - * or the timeout expires. + * @return mixed|false The pooled value, or false if the timeout expired. */ - public function pop(int $timeout): mixed + public function pop(float $timeout): mixed { - return $this->pool->pop($timeout); + // Swoole reads a non-positive channel timeout as "wait forever", the exact + // opposite of a zero budget, so clamp to a single short poll instead. + return $this->pool->pop($timeout > 0.0 ? $timeout : self::POLL); } public function count(): int @@ -53,16 +60,16 @@ public function count(): int * The lock is acquired before invoking the callback and is always released * afterward, even if the callback throws an exception. * - * @param callable $callback Callback to execute within the critical section. + * @param callable $callback Callback to execute within the critical section. * @return mixed The value returned by the callback. * * @throws \RuntimeException If the lock cannot be acquired within the timeout. - */ + */ public function synchronized(callable $callback): mixed { $acquired = $this->lock->lock(); - if (!$acquired) { + if (! $acquired) { throw new \RuntimeException('Failed to acquire lock'); } diff --git a/packages/pools/src/Pools/Connection.php b/packages/pools/src/Pools/Connection.php index 5e49b948d..36c0b322e 100644 --- a/packages/pools/src/Pools/Connection.php +++ b/packages/pools/src/Pools/Connection.php @@ -1,103 +1,44 @@ |null - */ - protected ?Pool $pool = null; - - /** - * @param TResource $resource - */ - public function __construct(protected mixed $resource) {} - - /** - * @return string - */ - public function getID(): string - { - return $this->id; - } - - /** - * @param string $id - * @return $this - */ - public function setID(string $id): static - { - $this->id = $id; - return $this; - } - /** - * @return TResource + * @param TResource $resource + * @param Pool $pool */ - public function getResource(): mixed - { - return $this->resource; - } - - /** - * @param TResource $resource - * @return $this - */ - public function setResource(mixed $resource): static - { - $this->resource = $resource; - return $this; - } + public function __construct( + public string $id, + public mixed $resource, + private Pool $pool, + ) {} /** - * @return Pool|null + * Return this connection to its pool for reuse. */ - public function getPool(): ?Pool + public function reclaim(): void { - return $this->pool; + $this->pool->reclaim($this); } /** - * @param Pool $pool - * @return $this + * Discard this connection and free its capacity for a replacement. */ - public function setPool(Pool $pool): static + public function destroy(): void { - $this->pool = $pool; - return $this; - } - - /** - * @return Pool - * @throws Exception - */ - public function reclaim(): Pool - { - if ($this->pool === null) { - throw new Exception('You cannot reclaim connection that does not have a pool.'); - } - - return $this->pool->reclaim($this); - } - - /** - * @return Pool - * @throws Exception - */ - public function destroy(): Pool - { - if ($this->pool === null) { - throw new Exception('You cannot destroy connection that does not have a pool.'); - } - - return $this->pool->destroy($this); + $this->pool->destroy($this); } } diff --git a/packages/pools/src/Pools/Group.php b/packages/pools/src/Pools/Group.php index b8175e4b0..9e5642a4f 100644 --- a/packages/pools/src/Pools/Group.php +++ b/packages/pools/src/Pools/Group.php @@ -1,9 +1,10 @@ $pool - * @return static + * @param Pool $pool */ public function add(Pool $pool): static { - $this->pools[$pool->getName()] = $pool; + $this->pools[$pool->name] = $pool; + return $this; } /** - * @param string $name * @return Pool + * * @throws Exception */ public function get(string $name): Pool @@ -32,19 +33,13 @@ public function get(string $name): Pool return $this->pools[$name] ?? throw new Exception("Pool '$name' not found"); } - /** - * @param string $name - * @return static - */ public function remove(string $name): static { unset($this->pools[$name]); + return $this; } - /** - * @return static - */ public function reclaim(): static { foreach ($this->pools as $pool) { @@ -58,20 +53,21 @@ public function reclaim(): static * Execute a callback with a managed connection * * @template TReturn - * @param array $names Name of resources - * @param callable(mixed...): TReturn $callback Function that receives the connection resources + * + * @param array $names Name of resources + * @param callable(mixed...): TReturn $callback Function that receives the connection resources * @return TReturn Return value from the callback + * * @throws Exception */ public function use(array $names, callable $callback): mixed { - if (empty($names)) { + if ($names === []) { throw new Exception('Cannot use with empty names'); } $connections = []; $pools = []; - $starts = []; $started = false; $failed = false; $thrown = null; @@ -80,13 +76,12 @@ public function use(array $names, callable $callback): mixed try { foreach ($names as $name) { $pool = $this->get($name); - $starts[] = microtime(true); $pools[] = $pool; $connections[] = $pool->pop(); } $started = true; - $result = $callback(...array_map(fn(Connection $connection) => $connection->getResource(), $connections)); + $result = $callback(...array_map(fn(Connection $connection): mixed => $connection->resource, $connections)); } catch (\Throwable $error) { $thrown = $error; $failed = $started; @@ -94,57 +89,22 @@ public function use(array $names, callable $callback): mixed $releaseError = null; - for ($i = \count($connections) - 1; $i >= 0; $i--) { + for ($i = \count($connections) - 1; $i >= 0; --$i) { try { - $pools[$i]->release($connections[$i], $failed, $starts[$i]); + $pools[$i]->release($connections[$i], $failed); } catch (\Throwable $error) { $releaseError ??= $error; } } - if ($thrown !== null) { + if ($thrown instanceof \Throwable) { throw $thrown; } - if ($releaseError !== null) { + if ($releaseError instanceof \Throwable) { throw $releaseError; } return $result; } - - /** - * @param int $reconnectAttempts - * @return static - */ - public function setReconnectAttempts(int $reconnectAttempts): static - { - foreach ($this->pools as $pool) { - $pool->setReconnectAttempts($reconnectAttempts); - } - - return $this; - } - - /** - * @param int $reconnectSleep - * @return static - */ - public function setReconnectSleep(int $reconnectSleep): static - { - foreach ($this->pools as $pool) { - $pool->setReconnectSleep($reconnectSleep); - } - - return $this; - } - - public function setTelemetry(Telemetry $telemetry): static - { - foreach ($this->pools as $pool) { - $pool->setTelemetry($telemetry); - } - - return $this; - } } diff --git a/packages/pools/src/Pools/Pool.php b/packages/pools/src/Pools/Pool.php index f57c756ee..d5869211c 100644 --- a/packages/pools/src/Pools/Pool.php +++ b/packages/pools/src/Pools/Pool.php @@ -1,220 +1,107 @@ > */ - protected array $active = []; - - /** - * Total number of connections created - */ - protected int $connectionsCreated = 0; - - private Histogram $telemetryWaitDuration; - private Histogram $telemetryUseDuration; - /** @var array */ - private array $telemetryAttributes; - - /** - * @param PoolAdapter $pool - * @param string $name - * @param int $size - * @param callable(): TResource $init - */ - public function __construct(protected PoolAdapter $pool, protected string $name, protected int $size, callable $init) - { - $this->init = $init; - // Initialize empty channel (no pre-filling for lazy initialization) - $this->pool->initialize($this->size); - $this->setTelemetry(new NoTelemetry()); - } - - /** - * @return string - */ - public function getName(): string - { - return $this->name; - } - - /** - * @return int - */ - public function getSize(): int - { - return $this->size; - } - - /** - * @return int - */ - public function getReconnectAttempts(): int - { - return $this->reconnectAttempts; - } - - /** - * @param int $reconnectAttempts - * @return $this - */ - public function setReconnectAttempts(int $reconnectAttempts): static - { - $this->reconnectAttempts = $reconnectAttempts; - return $this; - } - - /** - * @return int - */ - public function getReconnectSleep(): int - { - return $this->reconnectSleep; - } + private array $active = []; /** - * @param int $reconnectSleep - * @return $this + * Capacity accounted for: resources that exist, plus creations in flight. + * Never exceeds $size. */ - public function setReconnectSleep(int $reconnectSleep): static - { - $this->reconnectSleep = $reconnectSleep; - return $this; - } + private int $reserved = 0; /** - * @return int + * When each checked-out connection was requested, so the pool can measure + * use time itself instead of callers passing timestamps back in. + * + * @var array */ - public function getRetryAttempts(): int - { - return $this->retryAttempts; - } + private array $checkedOutAt = []; - /** - * @param int $retryAttempts - * @return $this - */ - public function setRetryAttempts(int $retryAttempts): static - { - $this->retryAttempts = $retryAttempts; - return $this; - } + private readonly Histogram $waitDuration; - /** - * @return int - */ - public function getRetrySleep(): int - { - return $this->retrySleep; - } + private readonly Histogram $useDuration; - /** - * @param int $retrySleep - * @return $this - */ - public function setRetrySleep(int $retrySleep): static - { - $this->retrySleep = $retrySleep; - return $this; - } + /** @var array */ + private readonly array $telemetryAttributes; + + /** + * @param PoolAdapter $adapter Storage and synchronisation for idle resources. + * @param string $name Identifies the pool in errors, connection ids and telemetry. + * @param int $size Maximum resources in existence at once. + * @param \Closure(): TResource $init Creates one resource. Own any retry policy here. + * @param float $timeout Seconds pop() may wait for an available connection. + * Excludes time spent inside $init, which the pool cannot bound. + */ + public function __construct( + private readonly PoolAdapter $adapter, + public readonly string $name, + public readonly int $size, + private readonly \Closure $init, + public readonly float $timeout, + ?Telemetry $telemetry = null, + ) { + if ($size < 1) { + throw new InvalidArgumentException("Pool '{$name}' size must be at least 1, got {$size}."); + } - /** - * Set the lock timeout for adapters that support synchronized locking. - * - * Note: - * - This setting is applied only if the underlying adapter supports lock timeouts. - * - For adapters that do not support locking or lock timeouts, this method is a no-op. - * - * @param int $timeout Synchronized lock timeout in seconds. - * @return $this - */ - public function setSynchronizationTimeout(int $timeout): static - { - $this->synchronizedTimeout = $timeout; - return $this; - } + if ($timeout < 0) { + throw new InvalidArgumentException("Pool '{$name}' timeout cannot be negative, got {$timeout}."); + } - public function getSynchronizationTimeout(): int - { - return $this->synchronizedTimeout; - } + // Start empty: resources are created on demand, never pre-filled. + $this->adapter->initialize($size); - /** - * @param Telemetry $telemetry - * @return $this - */ - public function setTelemetry(Telemetry $telemetry): static - { - $this->telemetryWaitDuration = Histogram::lazy( - telemetry: $telemetry, - name: 'pool.connection.wait_time', - unit: 's', - advisory: ['ExplicitBucketBoundaries' => [0.005, 0.01, 0.025, 0.05, 0.075, 0.1, 0.25, 0.5, 0.75, 1, 2.5, 5, 7.5, 10]], - ); - $this->telemetryUseDuration = Histogram::lazy( - telemetry: $telemetry, - name: 'pool.connection.use_time', - unit: 's', - advisory: ['ExplicitBucketBoundaries' => [0.005, 0.01, 0.025, 0.05, 0.075, 0.1, 0.25, 0.5, 0.75, 1, 2.5, 5, 7.5, 10]], - ); - $this->telemetryAttributes = ['pool' => $this->name, 'size' => $this->size]; + $telemetry ??= new NoTelemetry(); + $advisory = ['ExplicitBucketBoundaries' => [0.005, 0.01, 0.025, 0.05, 0.075, 0.1, 0.25, 0.5, 0.75, 1, 2.5, 5, 7.5, 10]]; + $this->waitDuration = Histogram::lazy(telemetry: $telemetry, name: 'pool.connection.wait_time', unit: 's', advisory: $advisory); + $this->useDuration = Histogram::lazy(telemetry: $telemetry, name: 'pool.connection.use_time', unit: 's', advisory: $advisory); + $this->telemetryAttributes = ['pool' => $name, 'size' => $size]; // Connection counts are gauges: only their value at export time matters, so observe // them lazily at collection rather than recording on every pop/push/reclaim. - $this->observeGauge($telemetry, 'pool.connection.active.count', fn() => \count($this->active)); - $this->observeGauge($telemetry, 'pool.connection.idle.count', fn() => $this->pool->count()); - $this->observeGauge($telemetry, 'pool.connection.open.count', fn() => \count($this->active) + $this->pool->count()); - $this->observeGauge($telemetry, 'pool.connection.capacity.count', fn() => $this->connectionsCreated); - - return $this; + $this->observeGauge($telemetry, 'pool.connection.active.count', fn(): int => \count($this->active)); + $this->observeGauge($telemetry, 'pool.connection.idle.count', fn(): int => $this->adapter->count()); + $this->observeGauge($telemetry, 'pool.connection.open.count', fn(): int => \count($this->active) + $this->adapter->count()); + $this->observeGauge($telemetry, 'pool.connection.capacity.count', fn(): int => $this->reserved); } /** * Register a connection-count observation on a (name-shared) gauge. * - * @param callable(): (float|int) $sample + * @param callable(): (float|int) $sample */ private function observeGauge(Telemetry $telemetry, string $name, callable $sample): void { @@ -223,44 +110,45 @@ private function observeGauge(Telemetry $telemetry, string $name, callable $samp } /** - * Execute a callback with a managed connection + * Execute a callback with a managed connection. + * + * The connection is returned to the pool afterwards, or discarded if the + * callback threw. This is the intended entry point; pop() and push() are + * exposed for callers that must manage the lifetime themselves. * * @template T - * @param callable(TResource): T $callback Function that receives the connection resource - * @return T Return value from the callback + * + * @param callable(TResource): T $callback Receives the connection resource. + * @return T */ public function use(callable $callback): mixed { - $start = microtime(true); $connection = null; $failed = false; try { $connection = $this->pop(); - return $callback($connection->getResource()); + + return $callback($connection->resource); } catch (\Throwable $error) { $failed = true; throw $error; } finally { - $this->telemetryUseDuration->record(microtime(true) - $start, $this->telemetryAttributes); - if ($connection !== null) { + if ($connection instanceof \Utopia\Pools\Connection) { $this->release($connection, $failed); } } } /** - * @param Connection $connection - * @return $this - * @internal + * Return a connection after use, reclaiming it or discarding it if the + * caller's work failed and the resource cannot be recovered. + * + * @param Connection $connection */ - public function release(Connection $connection, bool $failed = false, ?float $start = null): static + public function release(Connection $connection, bool $failed = false): static { - if ($start !== null) { - $this->telemetryUseDuration->record(microtime(true) - $start, $this->telemetryAttributes); - } - - if (!$failed) { + if (! $failed) { return $this->reclaim($connection); } @@ -287,20 +175,20 @@ public function release(Connection $connection, bool $failed = false, ?float $st * Last-resort cleanup when destroy() itself fails: the connection must never * stay tracked as active, or its slot is lost to the pool forever. * - * @param Connection $connection - * @return $this + * @param Connection $connection */ private function forget(Connection $connection): static { $untrack = function () use ($connection): void { - if (isset($this->active[$connection->getID()])) { - unset($this->active[$connection->getID()]); - $this->connectionsCreated--; + if (isset($this->active[$connection->id])) { + unset($this->active[$connection->id]); + unset($this->checkedOutAt[$connection->id]); + --$this->reserved; } }; try { - $this->pool->synchronized($untrack); + $this->adapter->synchronized($untrack); } catch (\Throwable) { $untrack(); } @@ -309,14 +197,16 @@ private function forget(Connection $connection): static } /** - * @param Connection $connection + * Ask a resource to make itself reusable after a failure. + * + * @param Connection $connection */ private function recover(Connection $connection): bool { - $resource = $connection->getResource(); + $resource = $connection->resource; - if (!\is_object($resource)) { - return !\is_resource($resource); + if (! \is_object($resource)) { + return ! \is_resource($resource); } try { @@ -343,220 +233,181 @@ private function recover(Connection $connection): bool } /** - * Summary: - * 1. Try to get a connection from the pool - * 2. If no connection is available, wait for one to be released - * 3. If still no connection is available, throw an exception - * 4. If a connection is available, return it + * Take a connection, creating one if there is spare capacity and otherwise + * waiting up to timeout for one to come back. * * @return Connection - * @throws Exception - * @internal Please migrate to `use`. + * + * @throws Exception When no connection becomes available within the budget. */ public function pop(): Connection { - $attempts = 0; - $totalSleepTime = 0; - $lastException = null; + $start = microtime(true); + $deadline = $start + $this->timeout; try { - do { - $attempts++; - // the connection creation block outside the lock so that other coroutines not get blocked in case of retries of a coroutine - // Lock: check + increment only - // Unlock - // Create connection (no lock) - // On failure: lock + decrement - $shouldCreateConnections = $this->pool->synchronized(function (): bool { - if ($this->pool->count() === 0 && $this->connectionsCreated < $this->size) { - $this->connectionsCreated++; - return true; - } - return false; - }); + $slot = $this->adapter->synchronized(function (): bool { + if ($this->adapter->count() === 0 && $this->reserved < $this->size) { + ++$this->reserved; - if ($shouldCreateConnections) { - $reserved = true; - - try { - $connection = $this->createConnection(); - $this->pool->synchronized(function () use ($connection): void { - $this->active[$connection->getID()] = $connection; - }); - $reserved = false; - - return $connection; - } catch (\Throwable $e) { - // Throwable, not Exception: the init callback is caller - // supplied, so an Error holds the slot just the same. - // Don't throw immediately - fall through to try getting - // an existing connection from the pool - $lastException = $e; - } finally { - if ($reserved) { - $this->pool->synchronized(function (): void { - $this->connectionsCreated--; - }); - } - } + return true; } - $connection = $this->pool->pop($this->getSynchronizationTimeout()); + return false; + }); - if ($connection === false || $connection === null) { - if ($attempts >= $this->getRetryAttempts()) { - $activeCount = \count($this->active); - $idleCount = $this->pool->count(); - $message = "Pool '{$this->name}' is empty (size {$this->size}, active {$activeCount}, idle {$idleCount})"; - throw new Exception($message, 0, $lastException); - } + if ($slot === true) { + // The slot is reserved before the resource exists, so every path out + // of this block has to release it or the capacity is lost for the + // lifetime of the process. + $handedOver = false; - $totalSleepTime += $this->getRetrySleep(); - sleep($this->getRetrySleep()); - } else { - if ($connection instanceof Connection) { - $this->pool->synchronized(function () use ($connection): void { - $this->active[$connection->getID()] = $connection; + try { + $connection = $this->createConnection(); + $this->track($connection, $start); + $handedOver = true; + + // Creation failures propagate untouched rather than falling + // through to a wait. Waiting for someone else's connection after + // our own create failed is a retry in disguise, and callers keep + // the original exception type to act on. + return $connection; + } finally { + if (! $handedOver) { + $this->adapter->synchronized(function (): void { + --$this->reserved; }); - return $connection; } } - } while ($attempts < $this->getRetryAttempts()); + } + + $connection = $this->adapter->pop(max(0.0, $deadline - microtime(true))); - $activeCount = \count($this->active); - $idleCount = $this->pool->count(); - throw new Exception("Pool '{$this->name}' failed to provide a connection (size {$this->size}, active {$activeCount}, idle {$idleCount})", 0, $lastException); + if ($connection instanceof Connection) { + $this->track($connection, $start); + + return $connection; + } + + throw new Exception(\sprintf( + "Pool '%s' could not provide a connection within %ss (size %d, active %d, idle %d)", + $this->name, + $this->timeout, + $this->size, + \count($this->active), + $this->adapter->count(), + )); } finally { - $this->telemetryWaitDuration->record($totalSleepTime, $this->telemetryAttributes); + $this->waitDuration->record(microtime(true) - $start, $this->telemetryAttributes); } } /** - * Create a new connection - * - * @return Connection - * @throws \Exception + * @param Connection $connection */ - protected function createConnection(): Connection + private function track(Connection $connection, float $requestedAt): void { - $connection = null; - $attempts = 0; - do { - try { - $attempts++; - $connection = new Connection(($this->init)()); - break; - } catch (\Exception $e) { - if ($attempts >= $this->getReconnectAttempts()) { - throw new \Exception('Failed to create connection: ' . $e->getMessage()); - } - sleep($this->getReconnectSleep()); - } - } while ($attempts < $this->getReconnectAttempts()); - - if ($connection === null) { - throw new \Exception('Failed to create connection'); - } + $this->adapter->synchronized(function () use ($connection, $requestedAt): void { + $this->active[$connection->id] = $connection; + $this->checkedOutAt[$connection->id] = $requestedAt; + }); + } - if (empty($connection->getID())) { - $connection->setID($this->getName() . '-' . uniqid()); + /** + * Record how long a connection was out, once, on whichever path returns it. + */ + private function recordUse(string $id): void + { + if (! isset($this->checkedOutAt[$id])) { + return; } - $connection->setPool($this); + $this->useDuration->record(microtime(true) - $this->checkedOutAt[$id], $this->telemetryAttributes); + unset($this->checkedOutAt[$id]); + } - return $connection; + /** + * @return Connection + */ + private function createConnection(): Connection + { + return new Connection($this->name . '-' . uniqid(), ($this->init)(), $this); } /** - * @param Connection $connection - * @return $this + * Hand a connection back to the idle set without inspecting it. + * + * @param Connection $connection */ public function push(Connection $connection): static { - // Push the actual connection back to the pool - $this->pool->push($connection); - unset($this->active[$connection->getID()]); + $this->recordUse($connection->id); + $this->adapter->push($connection); + unset($this->active[$connection->id]); return $this; } /** - * Returns the number of available connections (idle + not yet created) - * - * @return int + * Connections a caller could still obtain: idle plus not yet created. */ public function count(): int { - // Available = idle connections in pool + connections not yet created - return $this->pool->count() + ($this->size - $this->connectionsCreated); + return $this->adapter->count() + ($this->size - $this->reserved); } /** - * @param Connection|null $connection - * @return $this + * @param Connection|null $connection Reclaims every active connection when null. */ public function reclaim(?Connection $connection = null): static { - if ($connection !== null) { - $this->push($connection); - return $this; + if ($connection instanceof \Utopia\Pools\Connection) { + return $this->push($connection); } - foreach ($this->active as $connection) { - $this->push($connection); + foreach ($this->active as $active) { + $this->push($active); } return $this; } /** - * @param Connection|null $connection - * @return $this + * Discard a connection and free its capacity. The replacement is created + * lazily by the next pop(), so destroying never blocks on a connect and + * never fails for reasons unrelated to the connection being discarded. + * + * @param Connection|null $connection Destroys every active connection when null. */ - private function destroyConnection(?Connection $connection = null): static + public function destroy(?Connection $connection = null): static { - if ($connection !== null) { - $shouldCreate = $this->pool->synchronized(function () use ($connection) { - $this->connectionsCreated--; - unset($this->active[$connection->getID()]); - if ($this->connectionsCreated < $this->size) { - $this->connectionsCreated++; - return true; - } - return false; - }); - - if ($shouldCreate) { - try { - $this->pool->push($this->createConnection()); - } catch (\Throwable $e) { - $this->pool->synchronized(function (): void { - $this->connectionsCreated--; - }); - throw $e; - } + if (!$connection instanceof \Utopia\Pools\Connection) { + foreach (array_values($this->active) as $active) { + $this->destroy($active); } return $this; } - $activeConnections = array_values($this->active); - foreach ($activeConnections as $conn) { - $this->destroyConnection($conn); - } - return $this; - } - /** - * @param Connection|null $connection - * @return $this - */ - public function destroy(?Connection $connection = null): static - { - return $this->destroyConnection($connection); + $this->recordUse($connection->id); + + // Only release capacity this pool is actually holding. Destroying the same + // connection twice, or one belonging to another pool, must not drive + // reserved below the truth and let the pool exceed its size. + $this->adapter->synchronized(function () use ($connection): void { + if (! isset($this->active[$connection->id])) { + return; + } + + unset($this->active[$connection->id]); + --$this->reserved; + }); + + return $this; } /** - * @return bool + * No connection can be obtained without one being returned first. */ public function isEmpty(): bool { @@ -564,11 +415,10 @@ public function isEmpty(): bool } /** - * @return bool + * Every connection the pool could hand out is available. */ public function isFull(): bool { - // Pool is full when all possible connections are available (idle or not created yet) - return \count($this->active) === 0; + return $this->count() === $this->size; } } diff --git a/packages/pools/tests/Pools/Adapter/StackTest.php b/packages/pools/tests/Pools/Adapter/StackTest.php index bdd06190c..b3a3f93f7 100644 --- a/packages/pools/tests/Pools/Adapter/StackTest.php +++ b/packages/pools/tests/Pools/Adapter/StackTest.php @@ -1,11 +1,13 @@ setRetryAttempts(3); - $pool->setRetrySleep(0); - // Spawn 10 coroutines trying to get connections from a pool of 5 // First 5 should get connections immediately // Next 5 should wait and reuse connections after they're returned $channels = []; - for ($i = 0; $i < 10; $i++) { - $channels[$i] = new \Swoole\Coroutine\Channel(1); + for ($i = 0; $i < 10; ++$i) { + $channels[$i] = new Channel(1); } - for ($i = 0; $i < 10; $i++) { - \Swoole\Coroutine::create(function () use ($pool, $i, &$errors, &$successCount, $channels): void { + for ($i = 0; $i < 10; ++$i) { + Coroutine::create(function () use ($pool, $i, &$errors, &$successCount, $channels): void { try { // Each coroutine tries to get a connection $connection = $pool->pop(); - - if (empty($connection->getID())) { + if ($connection->id === '' || $connection->id === '0') { $errors[] = "Coroutine {$i}: Connection has no ID"; $channels[$i]->push(false); + return; } // Simulate some work - \Swoole\Coroutine::sleep(0.01); + Coroutine::sleep(0.01); // Return connection to pool $pool->reclaim($connection); - $successCount++; + ++$successCount; $channels[$i]->push(true); } catch (\Exception $e) { $errors[] = "Coroutine {$i}: " . $e->getMessage(); @@ -107,7 +108,7 @@ public function testSwooleCoroutineRaceCondition(): void public function testSwooleCoroutineHighConcurrency(): void { - if (!\extension_loaded('swoole')) { + if (! \extension_loaded('swoole')) { $this->markTestSkipped('Swoole extension is not loaded'); } @@ -118,31 +119,29 @@ public function testSwooleCoroutineHighConcurrency(): void \Swoole\Coroutine\run(function () use ($totalRequests, &$successCount, &$errorCount): void { // Create a pool with 3 connections inside coroutine context $connectionCounter = 0; - $pool = new Pool(new Swoole(), 'swoole-concurrent', 3, function () use (&$connectionCounter) { - $connectionCounter++; - return "connection-{$connectionCounter}"; - }); - - $pool->setRetryAttempts(3); - $pool->setRetrySleep(0); + $pool = new Pool(new Swoole(), 'swoole-concurrent', 3, function () use (&$connectionCounter): string { + ++$connectionCounter; + return "connection-{$connectionCounter}"; + }, timeout: 5.0); $channels = []; - for ($i = 0; $i < $totalRequests; $i++) { - $channels[$i] = new \Swoole\Coroutine\Channel(1); + for ($i = 0; $i < $totalRequests; ++$i) { + $channels[$i] = new Channel(1); } - for ($i = 0; $i < $totalRequests; $i++) { - \Swoole\Coroutine::create(function () use ($pool, $i, &$successCount, &$errorCount, $channels): void { + for ($i = 0; $i < $totalRequests; ++$i) { + Coroutine::create(function () use ($pool, $i, &$successCount, &$errorCount, $channels): void { try { - $pool->use(function ($resource) use ($i) { + $pool->use(function ($resource) use ($i): string { // Simulate work - \Swoole\Coroutine::sleep(0.01); + Coroutine::sleep(0.01); + return "processed-{$i}"; }); - $successCount++; + ++$successCount; $channels[$i]->push(true); } catch (\Exception) { - $errorCount++; + ++$errorCount; $channels[$i]->push(false); } }); @@ -167,7 +166,7 @@ public function testSwooleCoroutineHighConcurrency(): void public function testSwooleCoroutineConnectionUniqueness(): void { - if (!\extension_loaded('swoole')) { + if (! \extension_loaded('swoole')) { $this->markTestSkipped('Swoole extension is not loaded'); } @@ -177,25 +176,22 @@ public function testSwooleCoroutineConnectionUniqueness(): void \Swoole\Coroutine\run(function () use (&$seenResources, &$duplicateResources): void { // Create a pool with 5 connections inside coroutine context $connectionCounter = 0; - $pool = new Pool(new Swoole(), 'swoole-uniqueness', 5, function () use (&$connectionCounter) { - $connectionCounter++; - return "connection-{$connectionCounter}"; - }); - - $pool->setRetryAttempts(1); - $pool->setRetrySleep(0); + $pool = new Pool(new Swoole(), 'swoole-uniqueness', 5, function () use (&$connectionCounter): string { + ++$connectionCounter; + return "connection-{$connectionCounter}"; + }, timeout: 5.0); $channels = []; - for ($i = 0; $i < 5; $i++) { - $channels[$i] = new \Swoole\Coroutine\Channel(1); + for ($i = 0; $i < 5; ++$i) { + $channels[$i] = new Channel(1); } // Get all 5 connections simultaneously - for ($i = 0; $i < 5; $i++) { - \Swoole\Coroutine::create(function () use ($pool, $i, &$seenResources, &$duplicateResources, $channels): void { + for ($i = 0; $i < 5; ++$i) { + Coroutine::create(function () use ($pool, $i, &$seenResources, &$duplicateResources, $channels): void { try { $connection = $pool->pop(); - $resource = $connection->getResource(); + $resource = $connection->resource; // Check if we've seen this resource before (indicates race condition) if (isset($seenResources[$resource])) { @@ -205,7 +201,7 @@ public function testSwooleCoroutineConnectionUniqueness(): void } // Hold the connection briefly - \Swoole\Coroutine::sleep(0.01); + Coroutine::sleep(0.01); $channels[$i]->push(true); } catch (\Exception) { @@ -231,7 +227,7 @@ public function testSwooleCoroutineConnectionUniqueness(): void public function testSwooleCoroutineIdleConnectionReuse(): void { - if (!\extension_loaded('swoole')) { + if (! \extension_loaded('swoole')) { $this->markTestSkipped('Swoole extension is not loaded'); } @@ -240,20 +236,17 @@ public function testSwooleCoroutineIdleConnectionReuse(): void \Swoole\Coroutine\run(function () use (&$connectionIds, &$connectionCounter): void { // Create a pool with 3 connections inside coroutine context - $pool = new Pool(new Swoole(), 'swoole-reuse', 3, function () use (&$connectionCounter) { - $connectionCounter++; - return "connection-{$connectionCounter}"; - }); - - $pool->setRetryAttempts(1); - $pool->setRetrySleep(0); + $pool = new Pool(new Swoole(), 'swoole-reuse', 3, function () use (&$connectionCounter): string { + ++$connectionCounter; + return "connection-{$connectionCounter}"; + }, timeout: 5.0); // First wave: Create 3 connections $firstWave = []; - for ($i = 0; $i < 3; $i++) { + for ($i = 0; $i < 3; ++$i) { $conn = $pool->pop(); $firstWave[] = $conn; - $connectionIds['first'][] = $conn->getID(); + $connectionIds['first'][] = $conn->id; } // Return all connections @@ -263,10 +256,10 @@ public function testSwooleCoroutineIdleConnectionReuse(): void // Second wave: Should reuse the same 3 connections $secondWave = []; - for ($i = 0; $i < 3; $i++) { + for ($i = 0; $i < 3; ++$i) { $conn = $pool->pop(); $secondWave[] = $conn; - $connectionIds['second'][] = $conn->getID(); + $connectionIds['second'][] = $conn->id; } // Return all connections @@ -288,7 +281,7 @@ public function testSwooleCoroutineIdleConnectionReuse(): void public function testSwooleCoroutineStressTest(): void { - if (!\extension_loaded('swoole')) { + if (! \extension_loaded('swoole')) { $this->markTestSkipped('Swoole extension is not loaded'); } @@ -299,31 +292,29 @@ public function testSwooleCoroutineStressTest(): void \Swoole\Coroutine\run(function () use ($totalRequests, &$successCount, &$errorCount, &$connectionCounter): void { // Create a pool with 10 connections inside coroutine context - $pool = new Pool(new Swoole(), 'swoole-stress', 10, function () use (&$connectionCounter) { - $connectionCounter++; - return "connection-{$connectionCounter}"; - }); - - $pool->setRetryAttempts(10); - $pool->setRetrySleep(0); + $pool = new Pool(new Swoole(), 'swoole-stress', 10, function () use (&$connectionCounter): string { + ++$connectionCounter; + return "connection-{$connectionCounter}"; + }, timeout: 5.0); $channels = []; - for ($i = 0; $i < $totalRequests; $i++) { - $channels[$i] = new \Swoole\Coroutine\Channel(1); + for ($i = 0; $i < $totalRequests; ++$i) { + $channels[$i] = new Channel(1); } - for ($i = 0; $i < $totalRequests; $i++) { - \Swoole\Coroutine::create(function () use ($pool, $i, &$successCount, &$errorCount, $channels): void { + for ($i = 0; $i < $totalRequests; ++$i) { + Coroutine::create(function () use ($pool, $i, &$successCount, &$errorCount, $channels): void { try { $pool->use(function ($resource) { // Simulate variable work duration - \Swoole\Coroutine::sleep(0.001 * random_int(1, 5)); + Coroutine::sleep(0.001 * random_int(1, 5)); + return $resource; }); - $successCount++; + ++$successCount; $channels[$i]->push(true); } catch (\Exception) { - $errorCount++; + ++$errorCount; $channels[$i]->push(false); } }); @@ -341,9 +332,10 @@ public function testSwooleCoroutineStressTest(): void $this->assertSame(10, $pool->count(), 'Pool should have all connections back'); }); } - public function testInitOutsideCoroutineNotThrowAnyError(): void + + public function testInitOutsideCoroutineDoesNotThrow(): void { - $pool = new Pool(new Swoole(), 'test', 1, fn() => 'x'); + $pool = new Pool(new Swoole(), 'test', 1, fn(): string => 'x', timeout: 0.0); $this->assertInstanceOf(Pool::class, $pool); } } diff --git a/packages/pools/tests/Pools/Base.php b/packages/pools/tests/Pools/Base.php index 3acf0a665..5b3d42df3 100644 --- a/packages/pools/tests/Pools/Base.php +++ b/packages/pools/tests/Pools/Base.php @@ -1,5 +1,7 @@ + * A connection only exists as something checked out of a pool, so there is + * no standalone construction to test. + * + * @return Connection */ - protected Connection $connectionObject; - - protected function setUpConnection(): void - { - $this->connectionObject = new Connection('x'); - } - - public function testConnectionGetID(): void + private function checkedOutConnection(string $name = 'test'): Connection { - $this->execute(function (): void { - $this->setUpConnection(); - $this->assertSame('', $this->connectionObject->getID()); - - $this->connectionObject->setID('test'); - - $this->assertSame('test', $this->connectionObject->getID()); - }); + return new Pool($this->getAdapter(), $name, 2, fn(): string => 'x', timeout: 0.0)->pop(); } - public function testConnectionSetID(): void + public function testConnectionIdIsNamespacedByPool(): void { $this->execute(function (): void { - $this->setUpConnection(); - $this->assertSame('', $this->connectionObject->getID()); + $connection = $this->checkedOutConnection('alpha'); - $this->assertInstanceOf(Connection::class, $this->connectionObject->setID('test')); - - $this->assertSame('test', $this->connectionObject->getID()); - }); - } - - public function testConnectionGetResource(): void - { - $this->execute(function (): void { - $this->setUpConnection(); - $this->assertSame('x', $this->connectionObject->getResource()); + $this->assertStringStartsWith('alpha-', $connection->id); }); } - public function testConnectionSetResource(): void + public function testConnectionExposesItsResource(): void { $this->execute(function (): void { - $this->setUpConnection(); - $this->assertSame('x', $this->connectionObject->getResource()); - - $this->assertInstanceOf(Connection::class, $this->connectionObject->setResource('y')); - - $this->assertSame('y', $this->connectionObject->getResource()); - }); - } - - public function testConnectionSetPool(): void - { - $this->execute(function (): void { - $this->setUpConnection(); - $pool = new Pool($this->getAdapter(), 'test', 1, fn() => 'x'); - - $this->assertNull($this->connectionObject->getPool()); - $this->assertInstanceOf(Connection::class, $this->connectionObject->setPool($pool)); - }); - } - - public function testConnectionGetPool(): void - { - $this->execute(function (): void { - $this->setUpConnection(); - $pool = new Pool($this->getAdapter(), 'test', 1, fn() => 'x'); - - $this->assertNull($this->connectionObject->getPool()); - $this->assertInstanceOf(Connection::class, $this->connectionObject->setPool($pool)); - - $pool = $this->connectionObject->getPool(); - - if ($pool === null) { - throw new Exception('Pool should never be null here.'); - } - - $this->assertInstanceOf(Pool::class, $pool); - $this->assertSame('test', $pool->getName()); + $this->assertSame('x', $this->checkedOutConnection()->resource); }); } public function testConnectionReclaim(): void { $this->execute(function (): void { - $pool = new Pool($this->getAdapter(), 'test', 2, fn() => 'x'); + $pool = new Pool($this->getAdapter(), 'test', 2, fn(): string => 'x', timeout: 0.0); $this->assertSame(2, $pool->count()); @@ -111,33 +56,25 @@ public function testConnectionReclaim(): void $this->assertSame(0, $pool->count()); - $this->assertInstanceOf(Pool::class, $connection1->reclaim()); + $connection1->reclaim(); $this->assertSame(1, $pool->count()); - $this->assertInstanceOf(Pool::class, $connection2->reclaim()); + $connection2->reclaim(); $this->assertSame(2, $pool->count()); }); } - public function testConnectionReclaimException(): void - { - $this->execute(function (): void { - $this->setUpConnection(); - $this->expectException(Exception::class); - $this->connectionObject->reclaim(); - }); - } - public function testConnectionDestroy(): void { $this->execute(function (): void { $i = 0; - $object = new Pool($this->getAdapter(), 'testDestroy', 2, function () use (&$i) { - $i++; + $object = new Pool($this->getAdapter(), 'testDestroy', 2, function () use (&$i): string { + ++$i; + return $i <= 2 ? 'x' : 'y'; - }); + }, timeout: 0.0); $this->assertSame(2, $object->count()); @@ -146,12 +83,14 @@ public function testConnectionDestroy(): void $this->assertSame(0, $object->count()); - $this->assertSame('x', $connection1->getResource()); - $this->assertSame('x', $connection2->getResource()); + $this->assertSame('x', $connection1->resource); + $this->assertSame('x', $connection2->resource); $connection1->destroy(); $connection2->destroy(); + // Capacity is freed immediately; replacements are created lazily by the + // next pop() rather than inside destroy(). $this->assertSame(2, $object->count()); $connection1 = $object->pop(); @@ -159,8 +98,8 @@ public function testConnectionDestroy(): void $this->assertSame(0, $object->count()); - $this->assertSame('y', $connection1->getResource()); - $this->assertSame('y', $connection2->getResource()); + $this->assertSame('y', $connection1->resource); + $this->assertSame('y', $connection2->resource); }); } } diff --git a/packages/pools/tests/Pools/Scopes/GroupTestScope.php b/packages/pools/tests/Pools/Scopes/GroupTestScope.php index d21ad0282..7848ff9cf 100644 --- a/packages/pools/tests/Pools/Scopes/GroupTestScope.php +++ b/packages/pools/tests/Pools/Scopes/GroupTestScope.php @@ -1,8 +1,11 @@ execute(function (): void { $this->setUpGroup(); - $this->groupObject->add(new Pool($this->getAdapter(), 'test', 1, fn() => 'x')); + $this->groupObject->add(new Pool($this->getAdapter(), 'test', 1, fn(): string => 'x', timeout: 0.0)); $this->assertInstanceOf(Pool::class, $this->groupObject->get('test')); }); @@ -34,7 +38,7 @@ public function testGroupGet(): void { $this->execute(function (): void { $this->setUpGroup(); - $this->groupObject->add(new Pool($this->getAdapter(), 'test', 1, fn() => 'x')); + $this->groupObject->add(new Pool($this->getAdapter(), 'test', 1, fn(): string => 'x', timeout: 0.0)); $this->assertInstanceOf(Pool::class, $this->groupObject->get('test')); @@ -48,7 +52,7 @@ public function testGroupRemove(): void { $this->execute(function (): void { $this->setUpGroup(); - $this->groupObject->add(new Pool($this->getAdapter(), 'test', 1, fn() => 'x')); + $this->groupObject->add(new Pool($this->getAdapter(), 'test', 1, fn(): string => 'x', timeout: 0.0)); $this->assertInstanceOf(Pool::class, $this->groupObject->get('test')); @@ -64,7 +68,7 @@ public function testGroupReset(): void { $this->execute(function (): void { $this->setUpGroup(); - $this->groupObject->add(new Pool($this->getAdapter(), 'test', 5, fn() => 'x')); + $this->groupObject->add(new Pool($this->getAdapter(), 'test', 5, fn(): string => 'x', timeout: 0.0)); $this->assertSame(5, $this->groupObject->get('test')->count()); @@ -80,41 +84,13 @@ public function testGroupReset(): void }); } - public function testGroupReconnectAttempts(): void - { - $this->execute(function (): void { - $this->setUpGroup(); - $this->groupObject->add(new Pool($this->getAdapter(), 'test', 5, fn() => 'x')); - - $this->assertSame(3, $this->groupObject->get('test')->getReconnectAttempts()); - - $this->groupObject->setReconnectAttempts(5); - - $this->assertSame(5, $this->groupObject->get('test')->getReconnectAttempts()); - }); - } - - public function testGroupReconnectSleep(): void - { - $this->execute(function (): void { - $this->setUpGroup(); - $this->groupObject->add(new Pool($this->getAdapter(), 'test', 5, fn() => 'x')); - - $this->assertSame(1, $this->groupObject->get('test')->getReconnectSleep()); - - $this->groupObject->setReconnectSleep(2); - - $this->assertSame(2, $this->groupObject->get('test')->getReconnectSleep()); - }); - } - public function testGroupUse(): void { $this->execute(function (): void { $this->setUpGroup(); - $pool1 = new Pool($this->getAdapter(), 'pool1', 1, fn() => '1'); - $pool2 = new Pool($this->getAdapter(), 'pool2', 1, fn() => '2'); - $pool3 = new Pool($this->getAdapter(), 'pool3', 1, fn() => '3'); + $pool1 = new Pool($this->getAdapter(), 'pool1', 1, fn(): string => '1', timeout: 0.0); + $pool2 = new Pool($this->getAdapter(), 'pool2', 1, fn(): string => '2', timeout: 0.0); + $pool3 = new Pool($this->getAdapter(), 'pool3', 1, fn(): string => '3', timeout: 0.0); $this->groupObject->add($pool1); $this->groupObject->add($pool2); @@ -146,9 +122,9 @@ public function testGroupUseReclaimsEarlierConnectionWhenLaterPoolIsMissing(): v $this->setUpGroup(); $created = 0; $resources = []; - $pool = new Pool($this->getAdapter(), 'pool1', 1, function () use (&$created, &$resources) { - $created++; - $resources[] = new readonly class ('resource-' . $created) implements \Stringable { + $pool = new Pool($this->getAdapter(), 'pool1', 1, function () use (&$created, &$resources): object { + ++$created; + $resource = new readonly class ('resource-' . $created) implements \Stringable { public function __construct(private string $name) {} public function __toString(): string @@ -156,8 +132,10 @@ public function __toString(): string return $this->name; } }; - return $resources[$created - 1]; - }); + $resources[] = $resource; + + return $resource; + }, timeout: 0.0); $this->groupObject->add($pool); @@ -185,8 +163,7 @@ public function testGroupUseRecordsUseDurationTelemetry(): void $telemetry = new TestTelemetry(); $this->groupObject - ->add(new Pool($this->getAdapter(), 'pool1', 1, fn() => '1')) - ->setTelemetry($telemetry); + ->add(new Pool($this->getAdapter(), 'pool1', 1, fn(): string => '1', timeout: 0.0, telemetry: $telemetry)); $this->assertArrayNotHasKey('pool.connection.use_time', $telemetry->histograms); @@ -206,19 +183,20 @@ public function testGroupUseReleasesEveryConnectionWhenCleanupThrows(): void $this->execute(function (): void { $this->setUpGroup(); - $pool1 = new class ($this->getAdapter(), 'pool1', 1, fn() => '1') extends Pool { + $pool1 = new class ($this->getAdapter(), 'pool1', 1, fn(): string => '1', 0.0) extends Pool { public bool $released = false; - public function release(Connection $connection, bool $failed = false, ?float $start = null): static + public function release(Connection $connection, bool $failed = false): static { $this->released = true; - return parent::release($connection, $failed, $start); + + return parent::release($connection, $failed); } }; - $pool2 = new class ($this->getAdapter(), 'pool2', 1, fn() => '2') extends Pool { + $pool2 = new class ($this->getAdapter(), 'pool2', 1, fn(): string => '2', 0.0) extends Pool { public bool $released = false; - public function release(Connection $connection, bool $failed = false, ?float $start = null): static + public function release(Connection $connection, bool $failed = false): static { $this->released = true; throw new \RuntimeException('Release failed'); diff --git a/packages/pools/tests/Pools/Scopes/PoolTestScope.php b/packages/pools/tests/Pools/Scopes/PoolTestScope.php index 240ca2cd1..c5590fe0a 100644 --- a/packages/pools/tests/Pools/Scopes/PoolTestScope.php +++ b/packages/pools/tests/Pools/Scopes/PoolTestScope.php @@ -1,15 +1,20 @@ poolObject = new Pool($this->getAdapter(), 'test', 5, fn() => 'x'); + $this->poolObject = new Pool($this->getAdapter(), 'test', 5, fn(): string => 'x', timeout: 0.0); } public function testPoolGetName(): void { $this->execute(function (): void { $this->setUpPool(); - $this->assertSame('test', $this->poolObject->getName()); + $this->assertSame('test', $this->poolObject->name); }); } @@ -34,87 +39,7 @@ public function testPoolGetSize(): void { $this->execute(function (): void { $this->setUpPool(); - $this->assertSame(5, $this->poolObject->getSize()); - }); - } - - public function testPoolGetReconnectAttempts(): void - { - $this->execute(function (): void { - $this->setUpPool(); - $this->assertSame(3, $this->poolObject->getReconnectAttempts()); - }); - } - - public function testPoolSetReconnectAttempts(): void - { - $this->execute(function (): void { - $this->setUpPool(); - $this->assertSame(3, $this->poolObject->getReconnectAttempts()); - - $this->poolObject->setReconnectAttempts(20); - - $this->assertSame(20, $this->poolObject->getReconnectAttempts()); - }); - } - - public function testPoolGetReconnectSleep(): void - { - $this->execute(function (): void { - $this->setUpPool(); - $this->assertSame(1, $this->poolObject->getReconnectSleep()); - }); - } - - public function testPoolSetReconnectSleep(): void - { - $this->execute(function (): void { - $this->setUpPool(); - $this->assertSame(1, $this->poolObject->getReconnectSleep()); - - $this->poolObject->setReconnectSleep(20); - - $this->assertSame(20, $this->poolObject->getReconnectSleep()); - }); - } - - public function testPoolGetRetryAttempts(): void - { - $this->execute(function (): void { - $this->setUpPool(); - $this->assertSame(3, $this->poolObject->getRetryAttempts()); - }); - } - - public function testPoolSetRetryAttempts(): void - { - $this->execute(function (): void { - $this->setUpPool(); - $this->assertSame(3, $this->poolObject->getRetryAttempts()); - - $this->poolObject->setRetryAttempts(20); - - $this->assertSame(20, $this->poolObject->getRetryAttempts()); - }); - } - - public function testPoolGetRetrySleep(): void - { - $this->execute(function (): void { - $this->setUpPool(); - $this->assertSame(1, $this->poolObject->getRetrySleep()); - }); - } - - public function testPoolSetRetrySleep(): void - { - $this->execute(function (): void { - $this->setUpPool(); - $this->assertSame(1, $this->poolObject->getRetrySleep()); - - $this->poolObject->setRetrySleep(20); - - $this->assertSame(20, $this->poolObject->getRetrySleep()); + $this->assertSame(5, $this->poolObject->size); }); } @@ -129,7 +54,7 @@ public function testPoolPop(): void $this->assertSame(4, $this->poolObject->count()); $this->assertInstanceOf(Connection::class, $connection); - $this->assertSame('x', $connection->getResource()); + $this->assertSame('x', $connection->resource); // Pop remaining 4 connections $this->poolObject->pop(); @@ -167,7 +92,7 @@ public function testPoolPush(): void $this->assertSame(4, $this->poolObject->count()); $this->assertInstanceOf(Connection::class, $connection); - $this->assertSame('x', $connection->getResource()); + $this->assertSame('x', $connection->resource); $this->assertInstanceOf(Pool::class, $this->poolObject->push($connection)); @@ -259,40 +184,15 @@ public function testPoolIsFull(): void }); } - public function testPoolRetry(): void - { - $this->execute(function (): void { - $this->setUpPool(); - $this->poolObject->setReconnectAttempts(2); - $this->poolObject->setReconnectSleep(2); - - $this->poolObject->pop(); - $this->poolObject->pop(); - $this->poolObject->pop(); - $this->poolObject->pop(); - $this->poolObject->pop(); - - // Pool should be empty - $this->expectException(Exception::class); - - $timeStart = time(); - $this->poolObject->pop(); - $timeEnd = time(); - - $timeDiff = $timeEnd - $timeStart; - - $this->assertGreaterThanOrEqual(4, $timeDiff); - }); - } - public function testPoolDestroy(): void { $this->execute(function (): void { $i = 0; - $object = new Pool($this->getAdapter(), 'testDestroy', 2, function () use (&$i) { - $i++; + $object = new Pool($this->getAdapter(), 'testDestroy', 2, function () use (&$i): string { + ++$i; + return $i <= 2 ? 'x' : 'y'; - }); + }, timeout: 0.0); $this->assertSame(2, $object->count()); @@ -301,8 +201,8 @@ public function testPoolDestroy(): void $this->assertSame(0, $object->count()); - $this->assertSame('x', $connection1->getResource()); - $this->assertSame('x', $connection2->getResource()); + $this->assertSame('x', $connection1->resource); + $this->assertSame('x', $connection2->resource); $object->destroy(); @@ -313,31 +213,55 @@ public function testPoolDestroy(): void $this->assertSame(0, $object->count()); - $this->assertSame('y', $connection1->getResource()); - $this->assertSame('y', $connection2->getResource()); + $this->assertSame('y', $connection1->resource); + $this->assertSame('y', $connection2->resource); }); } - public function testPopRetriesAfterConnectionCreationFailure(): void + public function testPopWaitsAcquireTimeoutThenThrows(): void { $this->execute(function (): void { - $callCount = 0; - $pool = new Pool($this->getAdapter(), 'test-retry', 1, function () use (&$callCount) { - $callCount++; - if ($callCount <= 1) { - throw new \Exception('Connection failed'); - } - return 'x'; - }); - $pool->setReconnectAttempts(1); - $pool->setReconnectSleep(0); - $pool->setRetrySleep(0); + // timeout is the whole budget: one wait, no retry loop on top. + $pool = new Pool($this->getAdapter(), 'test-budget', 1, fn(): string => 'x', timeout: 0.25); + $pool->pop(); - // With the fix, pop() should retry after creation failure - // First attempt: createConnection fails (callCount=1), falls through - // Second attempt: createConnection succeeds (callCount=2) - $connection = $pool->pop(); - $this->assertSame('x', $connection->getResource()); + $start = microtime(true); + + try { + $pool->pop(); + $this->fail('Should have thrown'); + } catch (Exception $e) { + $this->assertStringContainsString('could not provide a connection', $e->getMessage()); + } + + // Generously bounded: the point is that it is one budget, not + // attempts x (wait + sleep) as it used to be. + $this->assertLessThan(3.0, microtime(true) - $start); + }); + } + + public function testCreationFailureSurfacesToTheCaller(): void + { + $this->execute(function (): void { + // Creation is no longer retried inside the pool, and a failed create + // does not fall through to a wait. The caller gets init's own + // exception, untouched, so it keeps the type it needs to act on. + $calls = 0; + $pool = new Pool($this->getAdapter(), 'test-create-fails', 1, function () use (&$calls): string { + ++$calls; + throw new Exception('connect refused'); + }, timeout: 0.0); + + try { + $pool->pop(); + $this->fail('Should have thrown'); + } catch (Exception $e) { + $this->assertSame('connect refused', $e->getMessage()); + $this->assertNull($e->getPrevious()); + } + + $this->assertSame(1, $calls); + $this->assertSame(1, $pool->count()); }); } @@ -349,24 +273,57 @@ public function testPopReleasesReservedSlotWhenCreationThrowsError(): void // failed pop at a time. $pool = new Pool($this->getAdapter(), 'test-error-leak', 2, function (): string { throw new \TypeError('Connection init failed'); - }); - $pool->setReconnectAttempts(1); - $pool->setReconnectSleep(0); - $pool->setRetryAttempts(1); - $pool->setRetrySleep(0); - + }, timeout: 0.0); // More attempts than slots, so a kept reservation shows up as lost - // capacity by the end. - for ($i = 0; $i < 5; $i++) { + // capacity by the end. Counted rather than using fail() inside the + // loop, so a missing throw cannot be confused with the caught one. + $thrown = 0; + for ($i = 0; $i < 5; ++$i) { try { $pool->pop(); - $this->fail('Should have thrown'); - } catch (Exception $e) { - $this->assertInstanceOf(\TypeError::class, $e->getPrevious()); + } catch (\Throwable $e) { + ++$thrown; + // An Error propagates as itself rather than being wrapped. + $this->assertInstanceOf(\TypeError::class, $e); } } + $this->assertSame(5, $thrown); + $this->assertSame(2, $pool->count()); + }); + } + + public function testDoubleDestroyDoesNotInflateCapacity(): void + { + $this->execute(function (): void { + // destroy() used to decrement reserved unconditionally, so destroying + // the same connection twice drove the count below the truth and let the + // pool create beyond its size. + $created = 0; + $pool = new Pool($this->getAdapter(), 'test-double-destroy', 2, function () use (&$created): string { + ++$created; + + return 'x'; + }, timeout: 0.0); + + $connection = $pool->pop(); + $pool->destroy($connection); + $pool->destroy($connection); + $this->assertSame(2, $pool->count()); + + $pool->pop(); + $pool->pop(); + + try { + $pool->pop(); + $this->fail('Should have thrown: the pool is at capacity'); + } catch (Exception) { + // expected + } + + // One before the destroys, two after, and never a third slot. + $this->assertSame(3, $created); }); } @@ -374,9 +331,6 @@ public function testPoolEmptyErrorIncludesActiveCount(): void { $this->execute(function (): void { $this->setUpPool(); // size 5 - $this->poolObject->setRetryAttempts(1); - $this->poolObject->setRetrySleep(0); - // Pop all 5 $this->poolObject->pop(); $this->poolObject->pop(); @@ -397,8 +351,9 @@ public function testUseDestroysConnectionWhenRecoveryFails(): void { $this->execute(function (): void { $created = 0; - $pool = new Pool($this->getAdapter(), 'test-destroy-on-error', 2, function () use (&$created) { - $created++; + $pool = new Pool($this->getAdapter(), 'test-destroy-on-error', 2, function () use (&$created): \Stringable { + ++$created; + return new readonly class ('resource-' . $created, $created === 1) implements \Stringable { public function __construct(private string $name, private bool $failRecovery) {} @@ -414,10 +369,7 @@ public function reconnect(): void } } }; - }); - $pool->setReconnectAttempts(1); - $pool->setReconnectSleep(0); - + }, timeout: 0.0); try { $pool->use(function (\Stringable $resource): void { $this->assertSame('resource-1', (string) $resource); @@ -439,8 +391,9 @@ public function testUseDestroysConnectionWhenRecoveryReturnsFalse(): void { $this->execute(function (): void { $created = 0; - $pool = new Pool($this->getAdapter(), 'test-destroy-on-false-recovery', 2, function () use (&$created) { - $created++; + $pool = new Pool($this->getAdapter(), 'test-destroy-on-false-recovery', 2, function () use (&$created): \Stringable { + ++$created; + return new readonly class ('resource-' . $created) implements \Stringable { public function __construct(private string $name) {} @@ -454,10 +407,7 @@ public function reconnect(): bool return false; } }; - }); - $pool->setReconnectAttempts(1); - $pool->setReconnectSleep(0); - + }, timeout: 0.0); try { $pool->use(function (\Stringable $resource): void { $this->assertSame('resource-1', (string) $resource); @@ -479,8 +429,9 @@ public function testUseRecoversAndReusesConnectionWhenRecoverySucceeds(): void { $this->execute(function (): void { $created = 0; - $pool = new Pool($this->getAdapter(), 'test-recover-and-reuse', 2, function () use (&$created) { - $created++; + $pool = new Pool($this->getAdapter(), 'test-recover-and-reuse', 2, function () use (&$created): \Stringable { + ++$created; + return new readonly class ('resource-' . $created) implements \Stringable { public function __construct(private string $name) {} @@ -494,10 +445,7 @@ public function reconnect(): bool return true; } }; - }); - $pool->setReconnectAttempts(1); - $pool->setReconnectSleep(0); - + }, timeout: 0.0); try { $pool->use(function (\Stringable $resource): void { $this->assertSame('resource-1', (string) $resource); @@ -518,8 +466,9 @@ public function testUseDestroysObjectConnectionWithoutRecoveryHooks(): void { $this->execute(function (): void { $created = 0; - $pool = new Pool($this->getAdapter(), 'test-destroy-without-recovery', 2, function () use (&$created) { - $created++; + $pool = new Pool($this->getAdapter(), 'test-destroy-without-recovery', 2, function () use (&$created): \Stringable { + ++$created; + return new readonly class ('resource-' . $created) implements \Stringable { public function __construct(private string $name) {} @@ -528,10 +477,7 @@ public function __toString(): string return $this->name; } }; - }); - $pool->setReconnectAttempts(1); - $pool->setReconnectSleep(0); - + }, timeout: 0.0); try { $pool->use(function (\Stringable $resource): void { $this->assertSame('resource-1', (string) $resource); @@ -554,7 +500,7 @@ public function testUseDestroysNativeResourceConnectionAfterCallbackFailure(): v $this->execute(function (): void { $created = 0; $pool = new Pool($this->getAdapter(), 'test-destroy-native-resource', 2, function () use (&$created) { - $created++; + ++$created; $resource = fopen('php://temp', 'r+'); if ($resource === false) { throw new \RuntimeException('Failed to open stream'); @@ -564,10 +510,7 @@ public function testUseDestroysNativeResourceConnectionAfterCallbackFailure(): v rewind($resource); return $resource; - }); - $pool->setReconnectAttempts(1); - $pool->setReconnectSleep(0); - + }, timeout: 0.0); try { $pool->use(function ($resource): void { $this->assertSame('resource-1', stream_get_contents($resource)); @@ -588,7 +531,7 @@ public function testUseDestroysNativeResourceConnectionAfterCallbackFailure(): v public function testUseForgetsConnectionWhenDestroyCleanupFails(): void { $this->execute(function (): void { - $adapter = new class extends \Utopia\Pools\Adapter\Stack { + $adapter = new class extends Stack { public bool $failSynchronized = false; public function synchronized(callable $callback): mixed @@ -603,8 +546,9 @@ public function synchronized(callable $callback): mixed }; $created = 0; - $pool = new Pool($adapter, 'test-forget-on-destroy-failure', 1, function () use (&$created) { - $created++; + $pool = new Pool($adapter, 'test-forget-on-destroy-failure', 1, function () use (&$created): \Stringable { + ++$created; + return new readonly class ('resource-' . $created) implements \Stringable { public function __construct(private string $name) {} @@ -613,10 +557,7 @@ public function __toString(): string return $this->name; } }; - }); - $pool->setReconnectAttempts(1); - $pool->setReconnectSleep(0); - + }, timeout: 0.0); try { $pool->use(function (\Stringable $resource) use ($adapter): void { $this->assertSame('resource-1', (string) $resource); @@ -638,8 +579,8 @@ public function testUsePreservesCallbackExceptionWhenReplacementFails(): void { $this->execute(function (): void { $created = 0; - $pool = new Pool($this->getAdapter(), 'test-preserve-callback-error', 1, function () use (&$created) { - $created++; + $pool = new Pool($this->getAdapter(), 'test-preserve-callback-error', 1, function () use (&$created): \Stringable { + ++$created; if ($created > 1) { throw new \TypeError('Replacement failed'); } @@ -657,10 +598,7 @@ public function reconnect(): never throw new \RuntimeException('Recovery failed'); } }; - }); - $pool->setReconnectAttempts(1); - $pool->setReconnectSleep(0); - + }, timeout: 0.0); $error = null; try { $pool->use(function (\Stringable $resource): void { @@ -679,9 +617,8 @@ public function reconnect(): never public function testPoolTelemetry(): void { $this->execute(function (): void { - $this->setUpPool(); $telemetry = new TestTelemetry(); - $this->poolObject->setTelemetry($telemetry); + $this->poolObject = new Pool($this->getAdapter(), 'test', 5, fn(): string => 'x', timeout: 0.0, telemetry: $telemetry); $this->assertArrayHasKey('pool.connection.open.count', $telemetry->observableGauges); $this->assertArrayHasKey('pool.connection.active.count', $telemetry->observableGauges); @@ -700,13 +637,14 @@ public function testPoolTelemetry(): void $value = $observed; }); } + return $value; }; $this->assertSame(5, $this->poolObject->count()); $connections = []; - for ($i = 0; $i < 3; $i++) { + for ($i = 0; $i < 3; ++$i) { $connections[] = $this->poolObject->pop(); } @@ -748,10 +686,8 @@ public function testMultiplePoolsShareGaugesButEmitDistinctSeries(): void // series; a single-callback gauge would drop all but the last pool to bind. $telemetry = new TestTelemetry(); - $alpha = new Pool($this->getAdapter(), 'alpha', 5, fn() => 'x'); - $beta = new Pool($this->getAdapter(), 'beta', 5, fn() => 'x'); - $alpha->setTelemetry($telemetry); - $beta->setTelemetry($telemetry); + $alpha = new Pool($this->getAdapter(), 'alpha', 5, fn(): string => 'x', timeout: 0.0, telemetry: $telemetry); + $beta = new Pool($this->getAdapter(), 'beta', 5, fn(): string => 'x', timeout: 0.0, telemetry: $telemetry); $alpha->pop(); $beta->pop(); @@ -763,11 +699,14 @@ public function testMultiplePoolsShareGaugesButEmitDistinctSeries(): void $series = []; foreach ($gauge->callbacks as $callback) { $callback(function (float|int $value, iterable $attributes = []) use (&$series): void { - $attrs = []; + $pool = null; foreach ($attributes as $key => $attr) { - $attrs[$key] = $attr; + if ($key === 'pool' && \is_string($attr)) { + $pool = $attr; + } } - $series[$attrs['pool']] = $value; + $this->assertIsString($pool); + $series[$pool] = $value; }); } @@ -778,9 +717,8 @@ public function testMultiplePoolsShareGaugesButEmitDistinctSeries(): void public function testPoolUseDurationTelemetryIsCreatedOnFirstUse(): void { $this->execute(function (): void { - $this->setUpPool(); $telemetry = new TestTelemetry(); - $this->poolObject->setTelemetry($telemetry); + $this->poolObject = new Pool($this->getAdapter(), 'test', 5, fn(): string => 'x', timeout: 0.0, telemetry: $telemetry); $this->assertArrayNotHasKey('pool.connection.use_time', $telemetry->histograms); diff --git a/packages/queue/composer.json b/packages/queue/composer.json index c4d200c65..4d15fef1c 100644 --- a/packages/queue/composer.json +++ b/packages/queue/composer.json @@ -37,7 +37,7 @@ "utopia-php/di": "^0.3", "utopia-php/lock": "^0.2", "utopia-php/servers": "^0.4", - "utopia-php/pools": "^1.0", + "utopia-php/pools": "^2.0", "utopia-php/telemetry": "^0.4", "utopia-php/validators": "0.3.*" }, diff --git a/packages/queue/tests/Queue/E2E/Adapter/PoolTest.php b/packages/queue/tests/Queue/E2E/Adapter/PoolTest.php index 325bbddd7..fb2163903 100644 --- a/packages/queue/tests/Queue/E2E/Adapter/PoolTest.php +++ b/packages/queue/tests/Queue/E2E/Adapter/PoolTest.php @@ -16,7 +16,7 @@ final class PoolTest extends Base { protected function getPublisher(): Publisher { - $pool = new UtopiaPool(new Stack(), 'redis', 1, fn(): \Utopia\Queue\Broker\Redis => new RedisBroker(new Redis('127.0.0.1', 16379), new Redis('127.0.0.1', 16379))); + $pool = new UtopiaPool(new Stack(), 'redis', 1, fn(): \Utopia\Queue\Broker\Redis => new RedisBroker(new Redis('127.0.0.1', 16379), new Redis('127.0.0.1', 16379)), timeout: 0.0); return new Pool($pool, $pool); }