From f343c9d01302094002092bd7a0b46127f948bb42 Mon Sep 17 00:00:00 2001 From: Chirag Aggarwal Date: Wed, 23 Sep 2026 09:30:03 +0100 Subject: [PATCH 1/2] chore: retire pools after its move to Appwrite utopia-php/pools now lives in appwrite/appwrite as packages/pools (appwrite/appwrite#13845). Drop the package here so this repository stops discovering and publishing it during Split runs, refresh the dependency graph, and point readers at the new source location. cache, client, messaging and queue keep their Packagist constraints on utopia-php/pools; the mirror and release tags remain available. --- README.md | 8 +- packages/pools/.github/workflows/mirror.yml | 17 - packages/pools/.gitignore | 16 - packages/pools/.gitpod.yml | 26 - packages/pools/LICENSE | 21 - packages/pools/README.md | 145 --- packages/pools/composer.json | 54 -- packages/pools/phpstan.neon | 7 - packages/pools/phpunit.xml | 12 - packages/pools/rector.php | 11 - packages/pools/src/Pools/Adapter.php | 58 -- packages/pools/src/Pools/Adapter/Stack.php | 70 -- packages/pools/src/Pools/Adapter/Swoole.php | 142 --- packages/pools/src/Pools/Connection.php | 64 -- packages/pools/src/Pools/Group.php | 110 --- packages/pools/src/Pools/Pool.php | 536 ----------- .../pools/tests/Pools/Adapter/StackTest.php | 21 - .../pools/tests/Pools/Adapter/SwooleTest.php | 341 ------- .../pools/tests/Pools/Adapter/WaiterTest.php | 218 ----- packages/pools/tests/Pools/Base.php | 26 - .../Pools/Scopes/ConnectionTestScope.php | 162 ---- .../tests/Pools/Scopes/GroupTestScope.php | 225 ----- .../tests/Pools/Scopes/PoolTestScope.php | 836 ------------------ .../tests/Pools/Scopes/TickingResource.php | 32 - 24 files changed, 2 insertions(+), 3156 deletions(-) delete mode 100644 packages/pools/.github/workflows/mirror.yml delete mode 100755 packages/pools/.gitignore delete mode 100644 packages/pools/.gitpod.yml delete mode 100644 packages/pools/LICENSE delete mode 100644 packages/pools/README.md delete mode 100755 packages/pools/composer.json delete mode 100644 packages/pools/phpstan.neon delete mode 100755 packages/pools/phpunit.xml delete mode 100644 packages/pools/rector.php delete mode 100644 packages/pools/src/Pools/Adapter.php delete mode 100644 packages/pools/src/Pools/Adapter/Stack.php delete mode 100644 packages/pools/src/Pools/Adapter/Swoole.php delete mode 100644 packages/pools/src/Pools/Connection.php delete mode 100644 packages/pools/src/Pools/Group.php delete mode 100644 packages/pools/src/Pools/Pool.php delete mode 100644 packages/pools/tests/Pools/Adapter/StackTest.php delete mode 100644 packages/pools/tests/Pools/Adapter/SwooleTest.php delete mode 100644 packages/pools/tests/Pools/Adapter/WaiterTest.php delete mode 100644 packages/pools/tests/Pools/Base.php delete mode 100644 packages/pools/tests/Pools/Scopes/ConnectionTestScope.php delete mode 100644 packages/pools/tests/Pools/Scopes/GroupTestScope.php delete mode 100644 packages/pools/tests/Pools/Scopes/PoolTestScope.php delete mode 100644 packages/pools/tests/Pools/Scopes/TickingResource.php diff --git a/README.md b/README.md index b69e60ab9..e3770cbf7 100644 --- a/README.md +++ b/README.md @@ -2,7 +2,7 @@ The source of truth for the [utopia-php](https://github.com/utopia-php) libraries. Each `packages/` is an independent Composer library; development happens here, and every push is mirrored to its read-only repository (e.g. `utopia-php/http`), so Composer/Packagist distribution is unchanged — mechanics in [docs/distribution.md](docs/distribution.md). -[`utopia-php/compression`](https://github.com/appwrite/appwrite/tree/main/packages/compression), [`utopia-php/smtp`](https://github.com/appwrite/appwrite/tree/main/packages/smtp), [`utopia-php/system`](https://github.com/appwrite/appwrite/tree/main/packages/system) and [`utopia-php/telemetry`](https://github.com/appwrite/appwrite/tree/main/packages/telemetry) are maintained in Appwrite. Their distribution mirrors and Packagist packages remain available. +[`utopia-php/compression`](https://github.com/appwrite/appwrite/tree/main/packages/compression), [`utopia-php/pools`](https://github.com/appwrite/appwrite/tree/main/packages/pools), [`utopia-php/smtp`](https://github.com/appwrite/appwrite/tree/main/packages/smtp), [`utopia-php/system`](https://github.com/appwrite/appwrite/tree/main/packages/system) and [`utopia-php/telemetry`](https://github.com/appwrite/appwrite/tree/main/packages/telemetry) are maintained in Appwrite. Their distribution mirrors and Packagist packages remain available. ## Quickstart @@ -46,17 +46,15 @@ To use an unmerged branch from an *external* consumer (e.g. test a fix in Appwri ## Dependency graph -Local siblings only: arrows point at dependencies between the packages in this checkout (`platform --> http` means platform requires http). Requirements on packages maintained elsewhere, such as the `utopia-php/telemetry` constraint that `cache`, `dns`, `http`, `messaging`, `pools`, `queue`, `schedule` and `storage` keep, are not drawn, so a node without arrows is not necessarily dependency-free. Regenerate with `bin/monorepo graph` after changing a package's requirements — `bin/monorepo validate` (which CI runs on every push) fails while it is stale. +Local siblings only: arrows point at dependencies between the packages in this checkout (`platform --> http` means platform requires http). Requirements on packages maintained elsewhere, such as the `utopia-php/telemetry` and `utopia-php/pools` constraints that `cache`, `client`, `dns`, `http`, `messaging`, `queue`, `schedule` and `storage` keep, are not drawn, so a node without arrows is not necessarily dependency-free. Regenerate with `bin/monorepo graph` after changing a package's requirements — `bin/monorepo validate` (which CI runs on every push) fails while it is stale. ```mermaid graph TD audit --> validators cache --> circuit-breaker - cache --> pools cdn --> client cli --> servers - client --> pools client --> span config --> validators console --> validators @@ -70,14 +68,12 @@ graph TD http --> servers http --> validators messaging --> client - messaging --> pools platform --> cli platform --> http platform --> queue platform --> servers queue --> di queue --> servers - queue --> pools queue --> validators servers --> di servers --> validators diff --git a/packages/pools/.github/workflows/mirror.yml b/packages/pools/.github/workflows/mirror.yml deleted file mode 100644 index a9a2c421b..000000000 --- a/packages/pools/.github/workflows/mirror.yml +++ /dev/null @@ -1,17 +0,0 @@ -name: Mirror - -on: - pull_request_target: - types: [opened] - issues: - types: [opened] - -permissions: - issues: write - pull-requests: write - -jobs: - redirect: - uses: utopia-php/monorepo/.github/workflows/mirror-redirect.yml@main - with: - package: pools diff --git a/packages/pools/.gitignore b/packages/pools/.gitignore deleted file mode 100755 index 9252d93c7..000000000 --- a/packages/pools/.gitignore +++ /dev/null @@ -1,16 +0,0 @@ -/vendor/ -/.idea/ -.DS_Store -mock.json -data-tests.php -loader.php -.phpunit.result.cache -/bin/view/results/ -.vscode -.vscode/* - -## - Oh Wess! -Makefile -.envrc -.vscode -composer.lock diff --git a/packages/pools/.gitpod.yml b/packages/pools/.gitpod.yml deleted file mode 100644 index 3f2315238..000000000 --- a/packages/pools/.gitpod.yml +++ /dev/null @@ -1,26 +0,0 @@ -tasks: - - name: Run Utopia Docker Stack - init: | - docker compose pull - docker compose build - command: | - docker run --rm --interactive --tty \ - --volume $PWD:/app \ - composer install \ - --ignore-platform-reqs \ - --optimize-autoloader \ - --no-plugins \ - --no-scripts \ - --prefer-dist - -vscode: - extensions: - - ms-azuretools.vscode-docker - -github: - # https://www.gitpod.io/docs/prebuilds#github-specific-configuration - prebuilds: - # enable for pull requests coming from forks (defaults to false) - pullRequestsFromForks: true - # add a check to pull requests (defaults to true) - addCheck: false diff --git a/packages/pools/LICENSE b/packages/pools/LICENSE deleted file mode 100644 index caa59ba60..000000000 --- a/packages/pools/LICENSE +++ /dev/null @@ -1,21 +0,0 @@ -MIT License - -Copyright (c) 2022 Utopia - -Permission is hereby granted, free of charge, to any person obtaining a copy -of this software and associated documentation files (the "Software"), to deal -in the Software without restriction, including without limitation the rights -to use, copy, modify, merge, publish, distribute, sublicense, and/or sell -copies of the Software, and to permit persons to whom the Software is -furnished to do so, subject to the following conditions: - -The above copyright notice and this permission notice shall be included in all -copies or substantial portions of the Software. - -THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR -IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, -FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE -AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER -LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, -OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE -SOFTWARE. diff --git a/packages/pools/README.md b/packages/pools/README.md deleted file mode 100644 index 41fea79a3..000000000 --- a/packages/pools/README.md +++ /dev/null @@ -1,145 +0,0 @@ -# Utopia Pools - -> [!IMPORTANT] -> This repository is a read-only mirror of the [utopia-php monorepo](https://github.com/utopia-php/monorepo). Development happens in [`packages/pools`](https://github.com/utopia-php/monorepo/tree/main/packages/pools) — please open issues and pull requests there. - -[![Build Status](https://travis-ci.com/utopia-php/pools.svg?branch=main)](https://travis-ci.com/utopia-php/pools) -![Total Downloads](https://img.shields.io/packagist/dt/utopia-php/pools.svg) -[![Discord](https://img.shields.io/discord/564160730845151244?label=discord)](https://appwrite.io/discord) - -Utopia pools library is simple and lite library for managing long living connection pools. This library is aiming to be as simple and easy to learn and use. This library is maintained by the [Appwrite team](https://appwrite.io). - -Although this library is part of the [Utopia Framework](https://github.com/utopia-php/framework) project it is dependency free, and can be used as standalone with any other PHP project or framework. - -## Concepts - -* **Pool** - A list of long living connections. You can pop connections out and use them and push them back to the pool for reuse. -* **Connection** - An object that holds a long living database or other external connection in a form of a resource. PDO object or a Redis client are examples of resources that can be used inside a connection. -* **Group** - A group of multiple pools. - -## Getting started - -Install using Composer: -```bash -composer require utopia-php/pools -``` -## Examples - -```php -use PDO; -use Utopia\Pools\Adapter\Swoole; -use Utopia\Pools\Group; -use Utopia\Pools\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')); -``` - -## Acquiring 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. - -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`. - -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`: - -```php -$pool = new Pool( - adapter: new Swoole, - name: 'mysql-pool', - size: 8, - init: fn () => $breaker->call(fn () => new PDO(/* ... */)), - timeout: 2.0, -); -``` - -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 pools requires PHP 8.4 or later. We recommend using the latest PHP version whenever possible. - -## Tests - -Run the test suite and static analysis from the monorepo root: - -```bash -bin/monorepo test pools -bin/monorepo check pools -``` - -## Copyright and license - -The MIT License (MIT) [http://www.opensource.org/licenses/mit-license.php](http://www.opensource.org/licenses/mit-license.php) diff --git a/packages/pools/composer.json b/packages/pools/composer.json deleted file mode 100755 index 4926ba46e..000000000 --- a/packages/pools/composer.json +++ /dev/null @@ -1,54 +0,0 @@ -{ - "name": "utopia-php/pools", - "description": "A simple library to manage connection pools", - "type": "library", - "keywords": [ - "php", - "framework", - "utopia", - "pools" - ], - "license": "MIT", - "minimum-stability": "stable", - "authors": [ - { - "name": "Team Appwrite", - "email": "team@appwrite.io" - } - ], - "autoload": { - "psr-4": { - "Utopia\\Pools\\": "src/Pools" - } - }, - "autoload-dev": { - "psr-4": { - "Utopia\\Tests\\": "tests/Pools" - } - }, - "scripts": { - "test": "phpunit --configuration phpunit.xml" - }, - "require": { - "php": ">=8.4", - "utopia-php/telemetry": "^0.4.6" - }, - "require-dev": { - "swoole/ide-helper": "6.*" - }, - "suggest": { - "ext-mongodb": "Needed to support MongoDB database pools", - "ext-redis": "Needed to support Redis cache pools", - "ext-pdo": "Needed to support MariaDB, MySQL or SQLite database pools", - "ext-swoole": "Needed to support Swoole based pool adapter" - }, - "config": { - "platform": { - "php": "8.4" - }, - "allow-plugins": { - "php-http/discovery": false, - "tbachert/spi": false - } - } -} diff --git a/packages/pools/phpstan.neon b/packages/pools/phpstan.neon deleted file mode 100644 index 68f401f1b..000000000 --- a/packages/pools/phpstan.neon +++ /dev/null @@ -1,7 +0,0 @@ -includes: - - ../../phpstan.neon - -parameters: - level: max - scanDirectories: - - vendor/swoole/ide-helper diff --git a/packages/pools/phpunit.xml b/packages/pools/phpunit.xml deleted file mode 100755 index 20e78338d..000000000 --- a/packages/pools/phpunit.xml +++ /dev/null @@ -1,12 +0,0 @@ - - - - ./tests/ - - - \ No newline at end of file diff --git a/packages/pools/rector.php b/packages/pools/rector.php deleted file mode 100644 index 35a483abe..000000000 --- a/packages/pools/rector.php +++ /dev/null @@ -1,11 +0,0 @@ -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 deleted file mode 100644 index b54398cc0..000000000 --- a/packages/pools/src/Pools/Adapter.php +++ /dev/null @@ -1,58 +0,0 @@ - */ - protected array $pool = []; - - /** - * Initialize the stack-based pool. - * - * Note: - * - `$size` is accepted for API compatibility with other pool adapters. - * - 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. - */ - public function initialize(int $size): static - { - $this->pool = []; - - return $this; - } - - public function push(mixed $connection): static - { - // Push connection to pool - $this->pool[] = $connection; - - return $this; - } - - /** - * Pop an item from the stack. - * - * Note: The stack adapter does not support blocking operations. - * The `$timeout` parameter is ignored. - * - * @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(float $timeout): mixed - { - return array_pop($this->pool); - } - - public function count(): int - { - return \count($this->pool); - } - - /** - * Executes the callback without acquiring a lock. - * - * This implementation does not provide mutual exclusion. - * - * @param callable $callback Callback to execute. - * @return mixed The value returned by the callback. - */ - public function synchronized(callable $callback): mixed - { - return $callback(); - } -} diff --git a/packages/pools/src/Pools/Adapter/Swoole.php b/packages/pools/src/Pools/Adapter/Swoole.php deleted file mode 100644 index 9a50d86fe..000000000 --- a/packages/pools/src/Pools/Adapter/Swoole.php +++ /dev/null @@ -1,142 +0,0 @@ - - */ - protected Channel $pool; - - protected Lock $lock; - - /** - * @var Channel Coalesced release signals. Kept apart from $pool so a - * signal is never mistaken for, or counted as, an idle resource. - */ - private Channel $signals; - - /** Shortest wait Swoole will honour without treating it as unbounded. */ - private const float POLL = 0.001; - - public function initialize(int $size): static - { - - $this->pool = new Channel($size); - $this->lock = new Lock(); - $this->signals = new Channel(1); - - return $this; - } - - public function push(mixed $connection): static - { - // Push connection to channel - $this->pool->push($connection); - // A returned resource changes what a parked caller is waiting on, which - // is exactly what unblock() is for. Waiters park on the signal rather - // than on $pool so one primitive releases them however the state moved. - $this->unblock(); - - return $this; - } - - /** - * Pop an item from the pool, waiting up to $timeout seconds. - * - * Returns as soon as a resource arrives OR the wait is released by - * {@see unblock()}; the caller decides what to do with an empty-handed - * return, which is why this reports only "here is one" or "not now". - * - * @return mixed|false The pooled value, or false if nothing was handed over. - */ - public function pop(float $timeout): mixed - { - // 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. - $budget = $timeout > 0.0 ? $timeout : self::POLL; - - if ($this->count() > 0) { - return $this->take(); - } - - // Park on the release signal rather than on the resource queue, so one - // primitive releases a caller however the state moved — a resource came - // back, or capacity was freed without one. - $this->signals->pop($budget); - - return $this->count() > 0 ? $this->take() : false; - } - - /** - * Take one idle resource and pass the baton on if more remain. - * - * Signals are coalesced to one, so a burst of returns that arrives while - * nobody is parked would otherwise release a single caller and leave the - * rest waiting beside resources that are already idle. - */ - private function take(): mixed - { - $connection = $this->pool->pop(self::POLL); - - if ($this->count() > 0) { - $this->unblock(); - } - - return $connection; - } - - #[\Override] - public function unblock(): static - { - // Coalesced, and remembered for a caller that has not parked yet: one - // signal is enough, because a released caller re-reads the real state. - if ($this->signals->length() === 0) { - $this->signals->push(true); - } - - return $this; - } - - /** - * Another coroutine can change the idle contents during a blocking pop(). - * @phpstan-impure - */ - public function count(): int - { - return (int) $this->pool->length(); - } - - /** - * Executes a callback while holding a lock. - * - * 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. - * @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) { - throw new \RuntimeException('Failed to acquire lock'); - } - - try { - return $callback(); - } finally { - $this->lock->unlock(); - } - } -} diff --git a/packages/pools/src/Pools/Connection.php b/packages/pools/src/Pools/Connection.php deleted file mode 100644 index 738c5404a..000000000 --- a/packages/pools/src/Pools/Connection.php +++ /dev/null @@ -1,64 +0,0 @@ -> - */ - private WeakReference $pool; - - /** - * @param TResource $resource - * @param Pool $pool - */ - public function __construct( - public string $id, - public mixed $resource, - Pool $pool, - ) { - $this->pool = WeakReference::create($pool); - } - - /** - * Return this connection to its pool for reuse. - */ - public function reclaim(): void - { - $this->pool->get()?->reclaim($this); - } - - /** - * Discard this connection and free its capacity for a replacement. - */ - public function destroy(): void - { - $this->pool->get()?->destroy($this); - } -} diff --git a/packages/pools/src/Pools/Group.php b/packages/pools/src/Pools/Group.php deleted file mode 100644 index 9e5642a4f..000000000 --- a/packages/pools/src/Pools/Group.php +++ /dev/null @@ -1,110 +0,0 @@ -> - */ - protected array $pools = []; - - /** - * @param Pool $pool - */ - public function add(Pool $pool): static - { - $this->pools[$pool->name] = $pool; - - return $this; - } - - /** - * @return Pool - * - * @throws Exception - */ - public function get(string $name): Pool - { - return $this->pools[$name] ?? throw new Exception("Pool '$name' not found"); - } - - public function remove(string $name): static - { - unset($this->pools[$name]); - - return $this; - } - - public function reclaim(): static - { - foreach ($this->pools as $pool) { - $pool->reclaim(); - } - - return $this; - } - - /** - * 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 - * @return TReturn Return value from the callback - * - * @throws Exception - */ - public function use(array $names, callable $callback): mixed - { - if ($names === []) { - throw new Exception('Cannot use with empty names'); - } - - $connections = []; - $pools = []; - $started = false; - $failed = false; - $thrown = null; - $result = null; - - try { - foreach ($names as $name) { - $pool = $this->get($name); - $pools[] = $pool; - $connections[] = $pool->pop(); - } - - $started = true; - $result = $callback(...array_map(fn(Connection $connection): mixed => $connection->resource, $connections)); - } catch (\Throwable $error) { - $thrown = $error; - $failed = $started; - } - - $releaseError = null; - - for ($i = \count($connections) - 1; $i >= 0; --$i) { - try { - $pools[$i]->release($connections[$i], $failed); - } catch (\Throwable $error) { - $releaseError ??= $error; - } - } - - if ($thrown instanceof \Throwable) { - throw $thrown; - } - - if ($releaseError instanceof \Throwable) { - throw $releaseError; - } - - return $result; - } -} diff --git a/packages/pools/src/Pools/Pool.php b/packages/pools/src/Pools/Pool.php deleted file mode 100644 index 33cdfd3b5..000000000 --- a/packages/pools/src/Pools/Pool.php +++ /dev/null @@ -1,536 +0,0 @@ -> - */ - private array $active = []; - - /** - * Capacity accounted for: resources that exist, plus creations in flight. - * Never exceeds $size. - */ - private int $reserved = 0; - - /** - * When each checked-out connection was requested, so the pool can measure - * use time itself instead of callers passing timestamps back in. - * - * @var array - */ - private array $checkedOutAt = []; - - private readonly Histogram $waitDuration; - - private readonly Histogram $useDuration; - - /** @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}."); - } - - if ($timeout < 0) { - throw new InvalidArgumentException("Pool '{$name}' timeout cannot be negative, got {$timeout}."); - } - - // Start empty: resources are created on demand, never pre-filled. - $this->adapter->initialize($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 = $telemetry->createHistogram(name: 'pool.connection.wait_time', unit: 's', advisory: $advisory); - $this->useDuration = $telemetry->createHistogram(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(): 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 - */ - private function observeGauge(Telemetry $telemetry, string $name, callable $sample): void - { - $telemetry->createObservableGauge($name) - ->observe(fn(callable $observe) => $observe($sample(), $this->telemetryAttributes)); - } - - /** - * 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 Receives the connection resource. - * @return T - */ - public function use(callable $callback): mixed - { - $connection = null; - $failed = false; - - try { - $connection = $this->pop(); - - return $callback($connection->resource); - } catch (\Throwable $error) { - $failed = true; - throw $error; - } finally { - if ($connection instanceof \Utopia\Pools\Connection) { - $this->release($connection, $failed); - } - } - } - - /** - * 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): static - { - if (! $failed) { - return $this->reclaim($connection); - } - - if ($this->recover($connection)) { - try { - return $this->reclaim($connection); - } catch (\Throwable) { - try { - return $this->destroy($connection); - } catch (\Throwable) { - return $this->forget($connection); - } - } - } - - try { - return $this->destroy($connection); - } catch (\Throwable) { - return $this->forget($connection); - } - } - - /** - * 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 - */ - private function forget(Connection $connection): static - { - $untrack = function () use ($connection): void { - if (isset($this->active[$connection->id])) { - unset($this->active[$connection->id]); - unset($this->checkedOutAt[$connection->id]); - --$this->reserved; - } - }; - - try { - $this->adapter->synchronized($untrack); - } catch (\Throwable) { - $untrack(); - } - $this->adapter->unblock(); - - return $this; - } - - /** - * Ask a resource to make itself reusable after a failure. - * - * @param Connection $connection - */ - private function recover(Connection $connection): bool - { - $resource = $connection->resource; - - if (! \is_object($resource)) { - return ! \is_resource($resource); - } - - try { - $recovered = false; - - if (method_exists($resource, 'reset')) { - $recovered = true; - if ($resource->reset() === false) { - return false; - } - } - - if (method_exists($resource, 'reconnect')) { - $recovered = true; - if ($resource->reconnect() === false) { - return false; - } - } - } catch (\Throwable) { - return false; - } - - return $recovered; - } - - /** - * Ask every idle resource to run its own upkeep. - * - * A pooled connection nobody checks out is not a connection nothing happens - * to: the server on the other end is still counting silence, and protocols - * that reap an idle peer (NATS closes after two missed pings) hand the next - * caller a socket that is already gone. The resource knows how to keep - * itself alive; what it lacks is a clock. The pool has no clock either, so - * a host that wants idle resources kept alive calls this on an interval - * shorter than the shortest server-side deadline it is holding. - * - * Duck-typed on `tick()` for the same reason {@see self::recover()} probes - * for `reset()`/`reconnect()`: pools depends on no driver, so the capability - * is discovered rather than declared. Resources without one are left alone, - * which is what makes this safe to call on every pool a host owns. - * - * Resources are taken and returned one at a time rather than under a single - * lock, because a tick writes to the network and holding the pool's lock - * across that would stall every pop() for the duration. The cost is that a - * resource checked out mid-sweep is simply not ticked this round — it has an - * active caller, which is the traffic the tick exists to substitute for. - */ - public function maintain(): static - { - /** @var list> $taken */ - $taken = []; - - try { - // Drain before ticking. Adapters are free to choose their own order - // — Stack is LIFO, Swoole's channel is FIFO — so popping and pushing - // one at a time hands the same resource straight back on a LIFO - // adapter and leaves every other one unswept. - for ($remaining = $this->adapter->count(); $remaining > 0; --$remaining) { - $connection = $this->adapter->pop(0.0); - - if (! $connection instanceof Connection) { - break; // a concurrent pop() drained the idle set ahead of us - } - - $taken[] = $connection; - } - - // foreach iterates a snapshot, so dropping each entry as it is dealt - // with leaves $taken holding exactly what the finally still owes. - foreach ($taken as $index => $connection) { - unset($taken[$index]); - - try { - $resource = $connection->resource; - - if (\is_object($resource) && method_exists($resource, 'tick')) { - $resource->tick(); - } - } catch (\Throwable) { - // A resource that cannot keep itself alive must not go back - // on the idle set: returning it hands the failure to the - // next caller as a mid-request error instead of a connect. - $this->releaseIdleSlot(); - - continue; - } - - // Returned as soon as its own tick is done rather than at the - // end, so the idle set refills while the sweep is still running. - $this->adapter->push($connection); - } - } finally { - // A resource stranded outside the pool is capacity lost for the - // life of the process, so an interrupted sweep still hands back - // everything it is holding. - foreach ($taken as $connection) { - $this->adapter->push($connection); - } - } - - return $this; - } - - /** - * Give back the capacity of a connection that was dropped from the idle set. - * - * {@see self::destroy()} cannot serve here: it releases capacity only for - * connections in $active, and an idle one by definition is not. The - * resource still exists though, so its slot is still counted in $reserved - * and has to be given back or the pool shrinks by one every sweep. - */ - private function releaseIdleSlot(): void - { - $this->adapter->synchronized(function (): void { - --$this->reserved; - }); - $this->adapter->unblock(); - } - - /** - * 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 When no connection becomes available within the budget. - */ - public function pop(): Connection - { - $start = microtime(true); - $deadline = $start + $this->timeout; - - try { - do { - $slot = $this->adapter->synchronized(function (): bool { - if ($this->adapter->count() === 0 && $this->reserved < $this->size) { - ++$this->reserved; - - return true; - } - - return false; - }); - - 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; - - 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; - }); - $this->adapter->unblock(); - } - } - } - - $connection = $this->adapter->pop(max(0.0, $deadline - microtime(true))); - - if ($connection instanceof Connection) { - $this->track($connection, $start); - - return $connection; - } - - // Empty-handed. That is not the same as out of time: capacity may - // have been freed by a caller that discarded its connection rather - // than returning one, and this acquisition can now create its own. - // Re-evaluate against the SAME deadline — the budget the caller - // asked for is never extended by looping. - if (microtime(true) >= $deadline) { - break; - } - } while (true); - - 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->waitDuration->record(microtime(true) - $start, $this->telemetryAttributes); - } - } - - /** - * @param Connection $connection - */ - private function track(Connection $connection, float $requestedAt): void - { - $this->adapter->synchronized(function () use ($connection, $requestedAt): void { - $this->active[$connection->id] = $connection; - $this->checkedOutAt[$connection->id] = $requestedAt; - }); - } - - /** - * 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; - } - - $this->useDuration->record(microtime(true) - $this->checkedOutAt[$id], $this->telemetryAttributes); - unset($this->checkedOutAt[$id]); - } - - /** - * @return Connection - */ - private function createConnection(): Connection - { - return new Connection($this->name . '-' . uniqid(), ($this->init)(), $this); - } - - /** - * Hand a connection back to the idle set without inspecting it. - * - * @param Connection $connection - */ - public function push(Connection $connection): static - { - $this->adapter->synchronized(function () use ($connection): void { - $this->recordUse($connection->id); - // A push can resume a waiting borrower inline. Its track() must wait - // until the previous checkout has been removed from active ownership. - $this->adapter->push($connection); - unset($this->active[$connection->id]); - }); - - return $this; - } - - /** - * Connections a caller could still obtain: idle plus not yet created. - */ - public function count(): int - { - return $this->adapter->count() + ($this->size - $this->reserved); - } - - /** - * @param Connection|null $connection Reclaims every active connection when null. - */ - public function reclaim(?Connection $connection = null): static - { - if ($connection instanceof \Utopia\Pools\Connection) { - return $this->push($connection); - } - - foreach ($this->active as $active) { - $this->push($active); - } - - 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. - */ - public function destroy(?Connection $connection = null): static - { - if (!$connection instanceof \Utopia\Pools\Connection) { - foreach (array_values($this->active) as $active) { - $this->destroy($active); - } - - return $this; - } - - $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; - }); - $this->adapter->unblock(); - - return $this; - } - - /** - * No connection can be obtained without one being returned first. - */ - public function isEmpty(): bool - { - return $this->count() === 0; - } - - /** - * Every connection the pool could hand out is available. - */ - public function isFull(): bool - { - return $this->count() === $this->size; - } -} diff --git a/packages/pools/tests/Pools/Adapter/StackTest.php b/packages/pools/tests/Pools/Adapter/StackTest.php deleted file mode 100644 index b3a3f93f7..000000000 --- a/packages/pools/tests/Pools/Adapter/StackTest.php +++ /dev/null @@ -1,21 +0,0 @@ -pop(); - - if ($connection->id === '' || $connection->id === '0') { - $errors[] = "Coroutine {$i}: Connection has no ID"; - $channels[$i]->push(false); - - return; - } - - // Simulate some work - Coroutine::sleep(0.01); - - // Return connection to pool - $pool->reclaim($connection); - - ++$successCount; - $channels[$i]->push(true); - } catch (\Exception $e) { - $errors[] = "Coroutine {$i}: " . $e->getMessage(); - $channels[$i]->push(false); - } - }); - } - - // Wait for all coroutines to complete - foreach ($channels as $channel) { - $channel->pop(); - } - - // Assertions inside coroutine context - $this->assertEmpty($errors, 'Errors occurred: ' . implode(', ', $errors)); - $this->assertSame(10, $successCount, 'All 10 coroutines should successfully complete'); - - // Pool should be full again after all connections are reclaimed - $this->assertSame(5, $pool->count(), 'Pool should have all 5 connections back'); - - // Should only create exactly pool size connections (no race conditions with new implementation) - $this->assertSame(5, $connectionCounter, 'Should create exactly 5 connections (pool size)'); - }); - } - - public function testSwooleCoroutineHighConcurrency(): void - { - if (! \extension_loaded('swoole')) { - $this->markTestSkipped('Swoole extension is not loaded'); - } - - $totalRequests = 20; - $successCount = 0; - $errorCount = 0; - - \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): string { - ++$connectionCounter; - - return "connection-{$connectionCounter}"; - }, timeout: 5.0); - $channels = []; - for ($i = 0; $i < $totalRequests; ++$i) { - $channels[$i] = new Channel(1); - } - - for ($i = 0; $i < $totalRequests; ++$i) { - Coroutine::create(function () use ($pool, $i, &$successCount, &$errorCount, $channels): void { - try { - $pool->use(function ($resource) use ($i): string { - // Simulate work - Coroutine::sleep(0.01); - - return "processed-{$i}"; - }); - ++$successCount; - $channels[$i]->push(true); - } catch (\Exception) { - ++$errorCount; - $channels[$i]->push(false); - } - }); - } - - // Wait for all coroutines to complete - foreach ($channels as $channel) { - $channel->pop(); - } - - // All requests should succeed with proper retry logic - $this->assertSame($totalRequests, $successCount, "All {$totalRequests} requests should succeed"); - $this->assertSame(0, $errorCount, 'No errors should occur with proper concurrency handling'); - - // Pool should be full again - $this->assertSame(3, $pool->count(), 'Pool should have all 3 connections back'); - - // Should only create 3 connections (pool size) - $this->assertSame(3, $connectionCounter, 'Should only create 3 connections (pool size)'); - }); - } - - public function testSwooleCoroutineConnectionUniqueness(): void - { - if (! \extension_loaded('swoole')) { - $this->markTestSkipped('Swoole extension is not loaded'); - } - - $seenResources = []; - $duplicateResources = []; - - \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): string { - ++$connectionCounter; - - return "connection-{$connectionCounter}"; - }, timeout: 5.0); - $channels = []; - for ($i = 0; $i < 5; ++$i) { - $channels[$i] = new Channel(1); - } - - // Get all 5 connections simultaneously - for ($i = 0; $i < 5; ++$i) { - Coroutine::create(function () use ($pool, $i, &$seenResources, &$duplicateResources, $channels): void { - try { - $connection = $pool->pop(); - $resource = $connection->resource; - - // Check if we've seen this resource before (indicates race condition) - if (isset($seenResources[$resource])) { - $duplicateResources[] = $resource; - } else { - $seenResources[$resource] = $connection; - } - - // Hold the connection briefly - Coroutine::sleep(0.01); - - $channels[$i]->push(true); - } catch (\Exception) { - $channels[$i]->push(false); - } - }); - } - - // Wait for all coroutines to complete - foreach ($channels as $channel) { - $channel->pop(); - } - - // Assertions inside coroutine context - $this->assertEmpty($duplicateResources, 'Duplicate resources detected: ' . implode(', ', $duplicateResources)); - $this->assertCount(5, $seenResources, 'Should have exactly 5 unique connections'); - - // Verify each connection has a unique resource - $resources = array_keys($seenResources); - $this->assertCount(5, array_unique($resources), 'All connection resources should be unique'); - }); - } - - public function testSwooleCoroutineIdleConnectionReuse(): void - { - if (! \extension_loaded('swoole')) { - $this->markTestSkipped('Swoole extension is not loaded'); - } - - $connectionIds = []; - $connectionCounter = 0; - - \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): string { - ++$connectionCounter; - - return "connection-{$connectionCounter}"; - }, timeout: 5.0); - // First wave: Create 3 connections - $firstWave = []; - for ($i = 0; $i < 3; ++$i) { - $conn = $pool->pop(); - $firstWave[] = $conn; - $connectionIds['first'][] = $conn->id; - } - - // Return all connections - foreach ($firstWave as $conn) { - $pool->reclaim($conn); - } - - // Second wave: Should reuse the same 3 connections - $secondWave = []; - for ($i = 0; $i < 3; ++$i) { - $conn = $pool->pop(); - $secondWave[] = $conn; - $connectionIds['second'][] = $conn->id; - } - - // Return all connections - foreach ($secondWave as $conn) { - $pool->reclaim($conn); - } - - // Assertions inside coroutine context - $this->assertSame(3, $connectionCounter, 'Should only create 3 connections total'); - $this->assertCount(3, $connectionIds['first'], 'First wave should have 3 connections'); - $this->assertCount(3, $connectionIds['second'], 'Second wave should have 3 connections'); - - // Second wave should reuse connections from first wave - sort($connectionIds['first']); - sort($connectionIds['second']); - $this->assertSame($connectionIds['first'], $connectionIds['second'], 'Second wave should reuse same connection IDs'); - }); - } - - public function testSwooleCoroutineStressTest(): void - { - if (! \extension_loaded('swoole')) { - $this->markTestSkipped('Swoole extension is not loaded'); - } - - $totalRequests = 100; - $successCount = 0; - $errorCount = 0; - $connectionCounter = 0; - - \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): string { - ++$connectionCounter; - - return "connection-{$connectionCounter}"; - }, timeout: 5.0); - $channels = []; - for ($i = 0; $i < $totalRequests; ++$i) { - $channels[$i] = new Channel(1); - } - - 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 - Coroutine::sleep(0.001 * random_int(1, 5)); - - return $resource; - }); - ++$successCount; - $channels[$i]->push(true); - } catch (\Exception) { - ++$errorCount; - $channels[$i]->push(false); - } - }); - } - - // Wait for all coroutines to complete - foreach ($channels as $channel) { - $channel->pop(); - } - - // Assertions inside coroutine context - $this->assertSame($totalRequests, $successCount, "All {$totalRequests} requests should succeed"); - $this->assertSame(0, $errorCount, 'No errors should occur'); - $this->assertSame(10, $connectionCounter, 'Should create exactly 10 connections (pool size)'); - $this->assertSame(10, $pool->count(), 'Pool should have all connections back'); - }); - } - - public function testInitOutsideCoroutineDoesNotThrow(): void - { - $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/Adapter/WaiterTest.php b/packages/pools/tests/Pools/Adapter/WaiterTest.php deleted file mode 100644 index d4906a6dd..000000000 --- a/packages/pools/tests/Pools/Adapter/WaiterTest.php +++ /dev/null @@ -1,218 +0,0 @@ -} $gauge */ - $gauge = $telemetry->observableGauges['pool.connection.active.count']; - foreach ($gauge->callbacks as $callback) { - $callback(static function (float|int $value) use (&$active): void { - $active = $value; - }); - } - return [$active, $adapter->count(), $pool->count()]; - }; - Coroutine\run(static function () use ($pool, $failure, $sample, &$results, &$snapshots, &$identities): void { - $first = $pool->pop(); - $identities[] = $first->resource; - $completed = new Channel(2); - $entered = new Channel(1); - Coroutine::create(static function () use ($pool, $failure, $completed, $entered, &$identities): void { - try { - $pool->use(static function (object $resource) use ($failure, $entered, &$identities): never { - $identities[] = $resource; - $entered->push(true); - Coroutine::sleep(0.02); - throw $failure; - }); - } catch (\Throwable $error) { - $completed->push($error); - } - }); - Coroutine::create(static function () use ($pool, $completed, &$identities): void { - try { - $completed->push($pool->use(static function (object $resource) use (&$identities): string { - $identities[] = $resource; - return 'replacement'; - })); - } catch (\Throwable $error) { - $completed->push($error); - } - }); - $pool->reclaim($first); - // Sample after the successor owns the checkout, including adapters - // that schedule it after the predecessor releases the bookkeeping lock. - $entered->pop(1); - $snapshots[] = $sample(); - $results = [$completed->pop(1), $completed->pop(1)]; - $snapshots[] = $sample(); - }); - $this->assertSame([1, 0, 0], $snapshots[0]); - $this->assertContains($failure, $results); - $this->assertContains('replacement', $results); - $this->assertSame(2, $created); - $this->assertCount(3, $identities); - $this->assertSame($identities[0], $identities[1]); - $this->assertNotSame($identities[1], $identities[2]); - $this->assertSame([0, 1, 1], $snapshots[1]); - } - - public function testDiscardingAFailedBorrowerWakesAnExistingWaiter(): void - { - $failure = new \RuntimeException('request failed'); - $results = []; - $created = 0; - $pool = new Pool(new Swoole(), 'discard', 1, static function () use (&$created): object { - ++$created; - return new \stdClass(); - }, 0.5); - Coroutine\run(static function () use ($pool, $failure, &$results): void { - $completed = new Channel(2); - Coroutine::create(static function () use ($pool, $failure, $completed): void { - try { - $pool->use(static function () use ($failure): never { - Coroutine::sleep(0.02); - throw $failure; - }); - } catch (\Throwable $error) { - $completed->push($error); - } - }); - Coroutine::create(static function () use ($pool, $completed): void { - try { - $completed->push($pool->use(static fn(): string => 'replacement')); - } catch (\Throwable $error) { - $completed->push($error); - } - }); - $results = [$completed->pop(1), $completed->pop(1)]; - }); - $this->assertContains($failure, $results); - $this->assertContains('replacement', $results); - $this->assertSame(2, $created); - $this->assertTrue($pool->isFull()); - } - - public function testFailedCreationWakesAnExistingWaiterWithoutRetryingTheCreator(): void - { - $failure = new \RuntimeException('connection failed'); - $created = 0; - $results = []; - $pool = new Pool(new Swoole(), 'creation', 1, static function () use (&$created, $failure): string { - if (++$created === 1) { - Coroutine::sleep(0.02); - throw $failure; - } - return 'replacement'; - }, 0.5); - Coroutine\run(static function () use ($pool, &$results): void { - $completed = new Channel(2); - for ($index = 0; $index < 2; ++$index) { - Coroutine::create(static function () use ($pool, $completed): void { - try { - $completed->push($pool->use(static fn(string $value): string => $value)); - } catch (\Throwable $error) { - $completed->push($error); - } - }); - } - $results = [$completed->pop(1), $completed->pop(1)]; - }); - $this->assertContains($failure, $results); - $this->assertContains('replacement', $results); - $this->assertSame(2, $created); - $this->assertTrue($pool->isFull()); - } - public function testReleasedWaitsDoNotExtendTheOriginalDeadline(): void - { - $adapter = new Swoole(); - $pool = new Pool($adapter, 'deadline', 1, static fn(): string => 'occupied', 0.04); - $failure = null; - $elapsed = 0.0; - Coroutine\run(static function () use ($pool, $adapter, &$failure, &$elapsed): void { - $connection = $pool->pop(); - $started = microtime(true); - Coroutine::create(static function () use ($adapter): void { - for ($index = 0; $index < 10; ++$index) { - Coroutine::sleep(0.01); - $adapter->unblock(); - } - }); - try { - $pool->pop(); - } catch (\Exception $error) { - $failure = $error; - } - $elapsed = microtime(true) - $started; - $pool->reclaim($connection); - }); - $this->assertInstanceOf(\Exception::class, $failure); - $this->assertLessThan(0.1, $elapsed); - $this->assertTrue($pool->isFull()); - } - - public function testSeveralWaitersCompeteForFreedCapacityWithoutOversubscribing(): void - { - $active = 0; - $maximum = 0; - $results = []; - $pool = new Pool(new Swoole(), 'capacity', 2, static fn(): object => new \stdClass(), 0.5); - Coroutine\run(static function () use ($pool, &$active, &$maximum, &$results): void { - $completed = new Channel(8); - for ($index = 0; $index < 8; ++$index) { - Coroutine::create(static function () use ($pool, $completed, &$active, &$maximum): void { - try { - $pool->use(static function () use (&$active, &$maximum): never { - ++$active; - $maximum = max($maximum, $active); - try { - Coroutine::sleep(0.002); - throw new \RuntimeException('discard'); - } finally { - --$active; - } - }); - } catch (\Throwable $error) { - $completed->push($error); - } - }); - } - for ($index = 0; $index < 8; ++$index) { - $results[] = $completed->pop(1); - } - }); - foreach ($results as $error) { - $this->assertInstanceOf(\RuntimeException::class, $error); - $this->assertSame('discard', $error->getMessage()); - } - $this->assertSame(2, $maximum); - $this->assertSame(0, $active); - $this->assertTrue($pool->isFull()); - } - -} diff --git a/packages/pools/tests/Pools/Base.php b/packages/pools/tests/Pools/Base.php deleted file mode 100644 index 5b3d42df3..000000000 --- a/packages/pools/tests/Pools/Base.php +++ /dev/null @@ -1,26 +0,0 @@ - - */ - private function checkedOutConnection(string $name = 'test'): Connection - { - return new Pool($this->getAdapter(), $name, 2, fn(): string => 'x', timeout: 0.0)->pop(); - } - - public function testConnectionIdIsNamespacedByPool(): void - { - $this->execute(function (): void { - $connection = $this->checkedOutConnection('alpha'); - - $this->assertStringStartsWith('alpha-', $connection->id); - }); - } - - public function testConnectionExposesItsResource(): void - { - $this->execute(function (): void { - $this->assertSame('x', $this->checkedOutConnection()->resource); - }); - } - - public function testConnectionReclaim(): void - { - $this->execute(function (): void { - $pool = new Pool($this->getAdapter(), 'test', 2, fn(): string => 'x', timeout: 0.0); - - $this->assertSame(2, $pool->count()); - - $connection1 = $pool->pop(); - - $this->assertSame(1, $pool->count()); - - $connection2 = $pool->pop(); - - $this->assertSame(0, $pool->count()); - - $connection1->reclaim(); - - $this->assertSame(1, $pool->count()); - - $connection2->reclaim(); - - $this->assertSame(2, $pool->count()); - }); - } - - public function testConnectionDestroy(): void - { - $this->execute(function (): void { - $i = 0; - $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()); - - $connection1 = $object->pop(); - $connection2 = $object->pop(); - - $this->assertSame(0, $object->count()); - - $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(); - $connection2 = $object->pop(); - - $this->assertSame(0, $object->count()); - - $this->assertSame('y', $connection1->resource); - $this->assertSame('y', $connection2->resource); - }); - } - - public function testDroppingAPoolFreesTheResourcesItWasHolding(): void - { - $this->execute(function (): void { - TrackedResource::$freed = 0; - - (function (): void { - $pool = new Pool($this->getAdapter(), 'lifetime', 3, fn(): TrackedResource => new TrackedResource(), timeout: 1.0); - - // Check every slot out at once so three distinct resources exist, - // then hand them all back: the pool is left holding them idle in - // its adapter's storage, which is where they used to become - // unreachable but uncollectable. - $connections = [$pool->pop(), $pool->pop(), $pool->pop()]; - - foreach ($connections as $connection) { - $connection->reclaim(); - } - })(); - - gc_collect_cycles(); - - // A strong connection-to-pool reference closes a cycle that the - // collector cannot trace, because with the Swoole adapter one edge - // runs through a Coroutine\Channel and lives in C memory. Every pool - // and pooled resource then survived for the life of the process. - $this->assertSame(3, TrackedResource::$freed); - }); - } - - public function testAConnectionOutlivesItsPoolWithoutReclaimingFailing(): void - { - $this->execute(function (): void { - // The pool is unreferenced the moment pop() returns, so a weakly held - // owner is already gone by the time the connection is handed back. - $connection = $this->checkedOutConnection('orphan'); - - gc_collect_cycles(); - - $connection->reclaim(); - $connection->destroy(); - - // Returning capacity to a pool that no longer exists is a no-op, but - // the connection still owns its resource. - $this->assertSame('x', $connection->resource); - }); - } -} - -final class TrackedResource -{ - public static int $freed = 0; - - public function __destruct() - { - ++self::$freed; - } -} diff --git a/packages/pools/tests/Pools/Scopes/GroupTestScope.php b/packages/pools/tests/Pools/Scopes/GroupTestScope.php deleted file mode 100644 index 7848ff9cf..000000000 --- a/packages/pools/tests/Pools/Scopes/GroupTestScope.php +++ /dev/null @@ -1,225 +0,0 @@ -groupObject = new Group(); - } - - public function testGroupAdd(): void - { - $this->execute(function (): void { - $this->setUpGroup(); - $this->groupObject->add(new Pool($this->getAdapter(), 'test', 1, fn(): string => 'x', timeout: 0.0)); - - $this->assertInstanceOf(Pool::class, $this->groupObject->get('test')); - }); - } - - public function testGroupGet(): void - { - $this->execute(function (): void { - $this->setUpGroup(); - $this->groupObject->add(new Pool($this->getAdapter(), 'test', 1, fn(): string => 'x', timeout: 0.0)); - - $this->assertInstanceOf(Pool::class, $this->groupObject->get('test')); - - $this->expectException(Exception::class); - - $this->assertInstanceOf(Pool::class, $this->groupObject->get('testx')); - }); - } - - public function testGroupRemove(): void - { - $this->execute(function (): void { - $this->setUpGroup(); - $this->groupObject->add(new Pool($this->getAdapter(), 'test', 1, fn(): string => 'x', timeout: 0.0)); - - $this->assertInstanceOf(Pool::class, $this->groupObject->get('test')); - - $this->groupObject->remove('test'); - - $this->expectException(Exception::class); - - $this->assertInstanceOf(Pool::class, $this->groupObject->get('test')); - }); - } - - public function testGroupReset(): void - { - $this->execute(function (): void { - $this->setUpGroup(); - $this->groupObject->add(new Pool($this->getAdapter(), 'test', 5, fn(): string => 'x', timeout: 0.0)); - - $this->assertSame(5, $this->groupObject->get('test')->count()); - - $this->groupObject->get('test')->pop(); - $this->groupObject->get('test')->pop(); - $this->groupObject->get('test')->pop(); - - $this->assertSame(2, $this->groupObject->get('test')->count()); - - $this->groupObject->reclaim(); - - $this->assertSame(5, $this->groupObject->get('test')->count()); - }); - } - - public function testGroupUse(): void - { - $this->execute(function (): void { - $this->setUpGroup(); - $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); - $this->groupObject->add($pool3); - - $this->assertSame(1, $pool1->count()); - $this->assertSame(1, $pool2->count()); - $this->assertSame(1, $pool3->count()); - - // @phpstan-ignore argument.type - $this->groupObject->use(['pool1', 'pool3'], function ($one, $three) use ($pool1, $pool2, $pool3): void { - $this->assertSame('1', $one); - $this->assertSame('3', $three); - - $this->assertSame(0, $pool1->count()); - $this->assertSame(1, $pool2->count()); - $this->assertSame(0, $pool3->count()); - }); - - $this->assertSame(1, $pool1->count()); - $this->assertSame(1, $pool2->count()); - $this->assertSame(1, $pool3->count()); - }); - } - - public function testGroupUseReclaimsEarlierConnectionWhenLaterPoolIsMissing(): void - { - $this->execute(function (): void { - $this->setUpGroup(); - $created = 0; - $resources = []; - $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 - { - return $this->name; - } - }; - $resources[] = $resource; - - return $resource; - }, timeout: 0.0); - - $this->groupObject->add($pool); - - try { - $this->groupObject->use(['pool1', 'missing'], function (): void {}); - $this->fail('Should have thrown'); - } catch (Exception) { - // expected - } - - $this->assertSame(1, $pool->count()); - - $pool->use(function (\Stringable $resource) use (&$resources): void { - $this->assertSame($resources[0], $resource); - $this->assertSame('resource-1', (string) $resource); - }); - $this->assertSame(1, $created); - }); - } - - public function testGroupUseRecordsUseDurationTelemetry(): void - { - $this->execute(function (): void { - $this->setUpGroup(); - $telemetry = new TestTelemetry(); - - $this->groupObject - ->add(new Pool($this->getAdapter(), 'pool1', 1, fn(): string => '1', timeout: 0.0, telemetry: $telemetry)); - - $this->assertArrayNotHasKey('pool.connection.use_time', $telemetry->histograms); - - $this->groupObject->use(['pool1'], function (...$resources): void { - $this->assertSame(['1'], $resources); - }); - - $this->assertArrayHasKey('pool.connection.use_time', $telemetry->histograms); - /** @var object{values: array} $useHistogram */ - $useHistogram = $telemetry->histograms['pool.connection.use_time']; - $this->assertCount(1, $useHistogram->values); - }); - } - - public function testGroupUseReleasesEveryConnectionWhenCleanupThrows(): void - { - $this->execute(function (): void { - $this->setUpGroup(); - - $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): static - { - $this->released = true; - - return parent::release($connection, $failed); - } - }; - $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): static - { - $this->released = true; - throw new \RuntimeException('Release failed'); - } - }; - - $this->groupObject - ->add($pool1) - ->add($pool2); - - $error = null; - try { - $this->groupObject->use(['pool1', 'pool2'], function (...$resources): void { - $this->assertSame(['1', '2'], $resources); - }); - } catch (\RuntimeException $error) { - } - - $this->assertInstanceOf(\RuntimeException::class, $error); - $this->assertSame('Release failed', $error->getMessage()); - $this->assertTrue($pool1->released); - $this->assertTrue($pool2->released); - $this->assertSame(1, $pool1->count()); - }); - } -} diff --git a/packages/pools/tests/Pools/Scopes/PoolTestScope.php b/packages/pools/tests/Pools/Scopes/PoolTestScope.php deleted file mode 100644 index 15e7b682c..000000000 --- a/packages/pools/tests/Pools/Scopes/PoolTestScope.php +++ /dev/null @@ -1,836 +0,0 @@ - - */ - protected Pool $poolObject; - - protected function setUpPool(): void - { - $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->name); - }); - } - - public function testPoolGetSize(): void - { - $this->execute(function (): void { - $this->setUpPool(); - $this->assertSame(5, $this->poolObject->size); - }); - } - - public function testPoolPop(): void - { - $this->execute(function (): void { - $this->setUpPool(); - $this->assertSame(5, $this->poolObject->count()); - - $connection = $this->poolObject->pop(); - - $this->assertSame(4, $this->poolObject->count()); - - $this->assertInstanceOf(Connection::class, $connection); - $this->assertSame('x', $connection->resource); - - // Pop remaining 4 connections - $this->poolObject->pop(); - $this->poolObject->pop(); - $this->poolObject->pop(); - $this->poolObject->pop(); - // Pool should be empty, next pop should throw - $this->expectException(Exception::class); - $this->poolObject->pop(); - }); - } - - public function testPoolUse(): void - { - $this->execute(function (): void { - $this->setUpPool(); - $this->assertSame(5, $this->poolObject->count()); - $this->poolObject->use(function ($resource): void { - $this->assertSame(4, $this->poolObject->count()); - $this->assertSame('x', $resource); - }); - - $this->assertSame(5, $this->poolObject->count()); - }); - } - - public function testPoolPush(): void - { - $this->execute(function (): void { - $this->setUpPool(); - $this->assertSame(5, $this->poolObject->count()); - - $connection = $this->poolObject->pop(); - - $this->assertSame(4, $this->poolObject->count()); - - $this->assertInstanceOf(Connection::class, $connection); - $this->assertSame('x', $connection->resource); - - $this->assertInstanceOf(Pool::class, $this->poolObject->push($connection)); - - $this->assertSame(5, $this->poolObject->count()); - }); - } - - public function testPoolCount(): void - { - $this->execute(function (): void { - $this->setUpPool(); - $this->assertSame(5, $this->poolObject->count()); - - $connection = $this->poolObject->pop(); - - $this->assertSame(4, $this->poolObject->count()); - - $this->poolObject->push($connection); - - $this->assertSame(5, $this->poolObject->count()); - }); - } - - public function testPoolReclaim(): void - { - $this->execute(function (): void { - $this->setUpPool(); - $this->assertSame(5, $this->poolObject->count()); - - $this->poolObject->pop(); - $this->poolObject->pop(); - $this->poolObject->pop(); - - $this->assertSame(2, $this->poolObject->count()); - - $this->poolObject->reclaim(); - - $this->assertSame(5, $this->poolObject->count()); - }); - } - - public function testPoolIsEmpty(): void - { - $this->execute(function (): void { - $this->setUpPool(); - $this->poolObject->pop(); - $this->poolObject->pop(); - $this->poolObject->pop(); - $this->poolObject->pop(); - $this->poolObject->pop(); - - $this->assertSame(true, $this->poolObject->isEmpty()); - }); - } - - public function testPoolIsFull(): void - { - $this->execute(function (): void { - $this->setUpPool(); - $this->assertSame(true, $this->poolObject->isFull()); - - $connection = $this->poolObject->pop(); - - $this->assertSame(false, $this->poolObject->isFull()); - - $this->poolObject->push($connection); - - $this->assertSame(true, $this->poolObject->isFull()); - - $this->poolObject->pop(); - $this->poolObject->pop(); - $this->poolObject->pop(); - $this->poolObject->pop(); - $this->poolObject->pop(); - - $this->assertSame(false, $this->poolObject->isFull()); - - $this->poolObject->reclaim(); - - $this->assertSame(true, $this->poolObject->isFull()); - - $this->poolObject->pop(); - $this->poolObject->pop(); - $this->poolObject->pop(); - $this->poolObject->pop(); - $this->poolObject->pop(); - - $this->assertSame(false, $this->poolObject->isFull()); - }); - } - - public function testPoolDestroy(): void - { - $this->execute(function (): void { - $i = 0; - $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()); - - $connection1 = $object->pop(); - $connection2 = $object->pop(); - - $this->assertSame(0, $object->count()); - - $this->assertSame('x', $connection1->resource); - $this->assertSame('x', $connection2->resource); - - $object->destroy(); - - $this->assertSame(2, $object->count()); - - $connection1 = $object->pop(); - $connection2 = $object->pop(); - - $this->assertSame(0, $object->count()); - - $this->assertSame('y', $connection1->resource); - $this->assertSame('y', $connection2->resource); - }); - } - - public function testPopWaitsAcquireTimeoutThenThrows(): void - { - $this->execute(function (): void { - // 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(); - - $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()); - }); - } - - public function testPopReleasesReservedSlotWhenCreationThrowsError(): void - { - $this->execute(function (): void { - // pop() reserves capacity before the connection exists. An Error - // escaping the catch used to keep that slot, draining the pool one - // failed pop at a time. - $pool = new Pool($this->getAdapter(), 'test-error-leak', 2, function (): string { - throw new \TypeError('Connection init failed'); - }, timeout: 0.0); - // More attempts than slots, so a kept reservation shows up as lost - // 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(); - } 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); - }); - } - - public function testPoolEmptyErrorIncludesActiveCount(): void - { - $this->execute(function (): void { - $this->setUpPool(); // size 5 - // Pop all 5 - $this->poolObject->pop(); - $this->poolObject->pop(); - $this->poolObject->pop(); - $this->poolObject->pop(); - $this->poolObject->pop(); - - try { - $this->poolObject->pop(); - $this->fail('Should have thrown'); - } catch (Exception $e) { - $this->assertStringContainsString('active 5', $e->getMessage()); - } - }); - } - - public function testUseDestroysConnectionWhenRecoveryFails(): void - { - $this->execute(function (): void { - $created = 0; - $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) {} - - public function __toString(): string - { - return $this->name; - } - - public function reconnect(): void - { - if ($this->failRecovery) { - throw new \RuntimeException('Recovery failed'); - } - } - }; - }, timeout: 0.0); - try { - $pool->use(function (\Stringable $resource): void { - $this->assertSame('resource-1', (string) $resource); - throw new \RuntimeException('Callback failed'); - }); - } catch (\RuntimeException) { - // expected - } - - $this->assertSame(2, $pool->count()); - - $pool->use(function (\Stringable $resource): void { - $this->assertSame('resource-2', (string) $resource); - }); - }); - } - - public function testUseDestroysConnectionWhenRecoveryReturnsFalse(): void - { - $this->execute(function (): void { - $created = 0; - $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) {} - - public function __toString(): string - { - return $this->name; - } - - public function reconnect(): bool - { - return false; - } - }; - }, timeout: 0.0); - try { - $pool->use(function (\Stringable $resource): void { - $this->assertSame('resource-1', (string) $resource); - throw new \RuntimeException('Callback failed'); - }); - } catch (\RuntimeException) { - // expected - } - - $this->assertSame(2, $pool->count()); - - $pool->use(function (\Stringable $resource): void { - $this->assertSame('resource-2', (string) $resource); - }); - }); - } - - public function testUseRecoversAndReusesConnectionWhenRecoverySucceeds(): void - { - $this->execute(function (): void { - $created = 0; - $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) {} - - public function __toString(): string - { - return $this->name; - } - - public function reconnect(): bool - { - return true; - } - }; - }, timeout: 0.0); - try { - $pool->use(function (\Stringable $resource): void { - $this->assertSame('resource-1', (string) $resource); - throw new \RuntimeException('Callback failed'); - }); - } catch (\RuntimeException) { - // expected - } - - $pool->use(function (\Stringable $resource) use (&$created): void { - $this->assertSame('resource-1', (string) $resource); - $this->assertSame(1, $created); - }); - }); - } - - public function testUseDestroysObjectConnectionWithoutRecoveryHooks(): void - { - $this->execute(function (): void { - $created = 0; - $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) {} - - public function __toString(): string - { - return $this->name; - } - }; - }, timeout: 0.0); - try { - $pool->use(function (\Stringable $resource): void { - $this->assertSame('resource-1', (string) $resource); - throw new \RuntimeException('Callback failed'); - }); - } catch (\RuntimeException) { - // expected - } - - $this->assertSame(2, $pool->count()); - - $pool->use(function (\Stringable $resource): void { - $this->assertSame('resource-2', (string) $resource); - }); - }); - } - - public function testUseDestroysNativeResourceConnectionAfterCallbackFailure(): void - { - $this->execute(function (): void { - $created = 0; - $pool = new Pool($this->getAdapter(), 'test-destroy-native-resource', 2, function () use (&$created) { - ++$created; - $resource = fopen('php://temp', 'r+'); - if ($resource === false) { - throw new \RuntimeException('Failed to open stream'); - } - - fwrite($resource, 'resource-' . $created); - rewind($resource); - - return $resource; - }, timeout: 0.0); - try { - $pool->use(function ($resource): void { - $this->assertSame('resource-1', stream_get_contents($resource)); - throw new \RuntimeException('Callback failed'); - }); - } catch (\RuntimeException) { - // expected - } - - $this->assertSame(2, $pool->count()); - - $pool->use(function ($resource): void { - $this->assertSame('resource-2', stream_get_contents($resource)); - }); - }); - } - - public function testUseForgetsConnectionWhenDestroyCleanupFails(): void - { - $this->execute(function (): void { - $adapter = new class extends Stack { - public bool $failSynchronized = false; - - public function synchronized(callable $callback): mixed - { - if ($this->failSynchronized) { - $this->failSynchronized = false; - throw new \RuntimeException('Lock failed'); - } - - return parent::synchronized($callback); - } - }; - - $created = 0; - $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) {} - - public function __toString(): string - { - return $this->name; - } - }; - }, timeout: 0.0); - try { - $pool->use(function (\Stringable $resource) use ($adapter): void { - $this->assertSame('resource-1', (string) $resource); - $adapter->failSynchronized = true; - throw new \RuntimeException('Callback failed'); - }); - } catch (\RuntimeException $exception) { - $this->assertSame('Callback failed', $exception->getMessage()); - } - - $pool->use(function (\Stringable $resource) use (&$created): void { - $this->assertSame('resource-2', (string) $resource); - $this->assertSame(2, $created); - }); - }); - } - - public function testUsePreservesCallbackExceptionWhenReplacementFails(): void - { - $this->execute(function (): void { - $created = 0; - $pool = new Pool($this->getAdapter(), 'test-preserve-callback-error', 1, function () use (&$created): \Stringable { - ++$created; - if ($created > 1) { - throw new \TypeError('Replacement failed'); - } - - return new readonly class ('resource-' . $created) implements \Stringable { - public function __construct(private string $name) {} - - public function __toString(): string - { - return $this->name; - } - - public function reconnect(): never - { - throw new \RuntimeException('Recovery failed'); - } - }; - }, timeout: 0.0); - $error = null; - try { - $pool->use(function (\Stringable $resource): void { - $this->assertSame('resource-1', (string) $resource); - throw new \LogicException('Callback failed'); - }); - } catch (\LogicException $error) { - } - - $this->assertInstanceOf(\LogicException::class, $error); - $this->assertSame('Callback failed', $error->getMessage()); - $this->assertSame(1, $pool->count()); - }); - } - - public function testPoolTelemetry(): void - { - $this->execute(function (): void { - $telemetry = new TestTelemetry(); - $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); - $this->assertArrayHasKey('pool.connection.idle.count', $telemetry->observableGauges); - $this->assertArrayHasKey('pool.connection.capacity.count', $telemetry->observableGauges); - $this->assertArrayNotHasKey('pool.connection.wait_time', $telemetry->histograms); - $this->assertArrayNotHasKey('pool.connection.use_time', $telemetry->histograms); - - // Observable gauges report their value at collection time, so read them on demand. - $read = function (string $name) use ($telemetry): float|int { - /** @var object{callbacks: array} $gauge */ - $gauge = $telemetry->observableGauges[$name]; - $value = 0; - foreach ($gauge->callbacks as $callback) { - $callback(function (float|int $observed) use (&$value): void { - $value = $observed; - }); - } - - return $value; - }; - - $this->assertSame(5, $this->poolObject->count()); - - $connections = []; - for ($i = 0; $i < 3; ++$i) { - $connections[] = $this->poolObject->pop(); - } - - $this->assertSame(3, $read('pool.connection.open.count')); - $this->assertSame(3, $read('pool.connection.active.count')); - $this->assertSame(0, $read('pool.connection.idle.count')); - $this->assertSame(3, $read('pool.connection.capacity.count')); - - /** @var object{values: array} $waitHistogram */ - $waitHistogram = $telemetry->histograms['pool.connection.wait_time']; - $this->assertCount(3, $waitHistogram->values); - $this->assertArrayNotHasKey('pool.connection.use_time', $telemetry->histograms); - - // Reclaim one connection: it returns to the pool as idle. - $this->poolObject->reclaim(array_pop($connections)); - - $this->assertSame(3, $read('pool.connection.open.count')); - $this->assertSame(2, $read('pool.connection.active.count')); - $this->assertSame(1, $read('pool.connection.idle.count')); - $this->assertSame(3, $read('pool.connection.capacity.count')); - - // Reclaim the rest. - foreach ($connections as $connection) { - $this->poolObject->reclaim($connection); - } - - $this->assertSame(3, $read('pool.connection.open.count')); - $this->assertSame(0, $read('pool.connection.active.count')); - $this->assertSame(3, $read('pool.connection.idle.count')); - $this->assertSame(5, $this->poolObject->count()); - }); - } - - public function testMultiplePoolsShareGaugesButEmitDistinctSeries(): void - { - $this->execute(function (): void { - // Adapters cache observable gauges by name, so every pool that registers - // 'pool.connection.*.count' shares one instrument. Each pool must still emit its own - // 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(): 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(); - $beta->pop(); - - /** @var object{callbacks: array} $gauge */ - $gauge = $telemetry->observableGauges['pool.connection.active.count']; - - $series = []; - foreach ($gauge->callbacks as $callback) { - $callback(function (float|int $value, iterable $attributes = []) use (&$series): void { - $pool = null; - foreach ($attributes as $key => $attr) { - if ($key === 'pool' && \is_string($attr)) { - $pool = $attr; - } - } - $this->assertIsString($pool); - $series[$pool] = $value; - }); - } - - $this->assertSame(['alpha' => 1, 'beta' => 2], $series); - }); - } - - public function testPoolUseDurationTelemetryIsCreatedOnFirstUse(): void - { - $this->execute(function (): void { - $telemetry = new TestTelemetry(); - $this->poolObject = new Pool($this->getAdapter(), 'test', 5, fn(): string => 'x', timeout: 0.0, telemetry: $telemetry); - - $this->assertArrayNotHasKey('pool.connection.use_time', $telemetry->histograms); - - $this->poolObject->use(function ($resource): void { - $this->assertSame('x', $resource); - }); - - $this->assertArrayHasKey('pool.connection.use_time', $telemetry->histograms); - /** @var object{values: array} $useHistogram */ - $useHistogram = $telemetry->histograms['pool.connection.use_time']; - $this->assertCount(1, $useHistogram->values); - }); - } - - /** - * A pool of resources that count their own ticks, so a sweep is observable. - * - * @param \Closure(): void|null $onTick Runs inside tick(), to fail one. - * @return Pool - */ - private function tickingPool(string $name, int $size, ?\Closure $onTick = null): Pool - { - $created = 0; - - return new Pool($this->getAdapter(), $name, $size, function () use (&$created, $onTick): TickingResource { - ++$created; - - return new TickingResource($created, $onTick); - }, timeout: 0.0); - } - - public function testMaintainTicksEveryIdleResource(): void - { - $this->execute(function (): void { - $pool = $this->tickingPool('test-maintain-ticks', 3); - - // Create three, then hand them all back so the whole set is idle. - $connections = [$pool->pop(), $pool->pop(), $pool->pop()]; - foreach ($connections as $connection) { - $pool->reclaim($connection); - } - - $pool->maintain(); - - foreach ($connections as $connection) { - $this->assertSame(1, $connection->resource->ticks); - } - - // The sweep is not a checkout: every resource is still available. - $this->assertSame(3, $pool->count()); - }); - } - - public function testMaintainLeavesCheckedOutResourcesUntouched(): void - { - $this->execute(function (): void { - $pool = $this->tickingPool('test-maintain-active', 2); - - $idle = $pool->pop(); - $active = $pool->pop(); - $pool->reclaim($idle); - - $pool->maintain(); - - // The active one has a caller driving it; that is the traffic the - // tick substitutes for, and ticking it would race that caller. - $this->assertSame(1, $idle->resource->ticks); - $this->assertSame(0, $active->resource->ticks); - }); - } - - public function testMaintainIgnoresResourcesThatCannotTick(): void - { - $this->execute(function (): void { - $pool = new Pool($this->getAdapter(), 'test-maintain-no-tick', 2, fn(): string => 'x', timeout: 0.0); - - $connection = $pool->pop(); - $pool->reclaim($connection); - - // A string resource has no tick(); the sweep must be a no-op rather - // than an error, so a host can call it on every pool it owns. - $pool->maintain(); - - $this->assertSame(2, $pool->count()); - $pool->use(function ($resource): void { - $this->assertSame('x', $resource); - }); - }); - } - - public function testMaintainDiscardsResourceWhoseTickThrows(): void - { - $this->execute(function (): void { - $pool = $this->tickingPool('test-maintain-discard', 2, function (): void { - throw new \RuntimeException('server closed the connection'); - }); - - $dead = $pool->pop(); - $pool->reclaim($dead); - - $pool->maintain(); - - // The dead resource is gone but its slot is not: the pool must not - // shrink by one on every sweep. - $this->assertSame(2, $pool->count()); - - // And the next caller gets a freshly created resource, not the one - // that just failed to keep itself alive. - $deadSerial = $dead->resource->serial; - $pool->use(function (TickingResource $resource) use ($deadSerial): void { - $this->assertNotSame($deadSerial, $resource->serial); - }); - }); - } -} diff --git a/packages/pools/tests/Pools/Scopes/TickingResource.php b/packages/pools/tests/Pools/Scopes/TickingResource.php deleted file mode 100644 index 81df3aa79..000000000 --- a/packages/pools/tests/Pools/Scopes/TickingResource.php +++ /dev/null @@ -1,32 +0,0 @@ -ticks; - - if ($this->onTick instanceof \Closure) { - ($this->onTick)(); - } - } -} From 788bcf8f458ab33b2caa339b4166bda8e8ceed03 Mon Sep 17 00:00:00 2001 From: Chirag Aggarwal Date: Wed, 23 Sep 2026 09:34:22 +0100 Subject: [PATCH 2/2] docs: list telemetry and pools dependants separately client requires only pools and dns, http, schedule and storage only telemetry, so one shared list overstated both. --- README.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/README.md b/README.md index e3770cbf7..270bb74e8 100644 --- a/README.md +++ b/README.md @@ -46,7 +46,7 @@ To use an unmerged branch from an *external* consumer (e.g. test a fix in Appwri ## Dependency graph -Local siblings only: arrows point at dependencies between the packages in this checkout (`platform --> http` means platform requires http). Requirements on packages maintained elsewhere, such as the `utopia-php/telemetry` and `utopia-php/pools` constraints that `cache`, `client`, `dns`, `http`, `messaging`, `queue`, `schedule` and `storage` keep, are not drawn, so a node without arrows is not necessarily dependency-free. Regenerate with `bin/monorepo graph` after changing a package's requirements — `bin/monorepo validate` (which CI runs on every push) fails while it is stale. +Local siblings only: arrows point at dependencies between the packages in this checkout (`platform --> http` means platform requires http). Requirements on packages maintained elsewhere, such as `utopia-php/telemetry` (required by `cache`, `dns`, `http`, `messaging`, `queue`, `schedule` and `storage`) and `utopia-php/pools` (required by `cache`, `client`, `messaging` and `queue`), are not drawn, so a node without arrows is not necessarily dependency-free. Regenerate with `bin/monorepo graph` after changing a package's requirements — `bin/monorepo validate` (which CI runs on every push) fails while it is stale. ```mermaid