From 2e82f0c2ece473cc96cf76c324a7e11bfa59a66d Mon Sep 17 00:00:00 2001 From: ArnabChatterjee20k Date: Tue, 30 Dec 2025 13:03:59 +0530 Subject: [PATCH] fix: update coroutine pool handling and configuration --- .env | 3 ++- app/init/registers.php | 41 +++++++++++++++++------------------------ app/realtime.php | 10 +++++----- docker-compose.yml | 1 + 4 files changed, 25 insertions(+), 30 deletions(-) diff --git a/.env b/.env index e849e83801..ff989ec9e1 100644 --- a/.env +++ b/.env @@ -125,4 +125,5 @@ _APP_WEBHOOK_MAX_FAILED_ATTEMPTS=10 _APP_PROJECT_REGIONS=default _APP_FUNCTIONS_CREATION_ABUSE_LIMIT=5000 _APP_STATS_USAGE_DUAL_WRITING_DBS=database_db_main -_APP_TRUSTED_HEADERS=x-forwarded-for \ No newline at end of file +_APP_TRUSTED_HEADERS=x-forwarded-for +COROUTINE_POOLS=disabled \ No newline at end of file diff --git a/app/init/registers.php b/app/init/registers.php index c5bba06df8..ab88d5fbbd 100644 --- a/app/init/registers.php +++ b/app/init/registers.php @@ -23,7 +23,7 @@ use Utopia\Logger\Adapter\LogOwl; use Utopia\Logger\Adapter\Raygun; use Utopia\Logger\Adapter\Sentry; use Utopia\Logger\Logger; -use Utopia\Pools\Adapter\Stack as Stack; +use Utopia\Pools\Adapter\Stack as StackPool; use Utopia\Pools\Adapter\Swoole as SwoolePool; use Utopia\Pools\Group; use Utopia\Pools\Pool; @@ -145,14 +145,7 @@ $register->set('realtimeLogger', function () { return new Logger($adapter); }); -/** - * Build a pool Group with shared config. - * - * @param string $configPrefix Config param prefix (e.g. 'pools', 'coroutinepools') - * @param callable(): \Utopia\Pools\Adapter $adapterFactory Factory returning the Pool adapter (Stack or Swoole) - * @param int|null $syncTimeout Optional synchronization timeout to apply on each pool (null to skip) - */ -$buildPoolGroup = function (string $configPrefix, callable $adapterFactory, ?int $syncTimeout = null): Group { +$register->set('pools', function () { $group = new Group(); $fallbackForDB = 'db_main=' . AppwriteURL::unparse([ @@ -231,7 +224,7 @@ $buildPoolGroup = function (string $configPrefix, callable $adapterFactory, ?int throw new \Exception('Pool size is too small. Increase the number of allowed database connections or decrease the number of workers.', 500); } - $poolSize = (int)(($instanceConnections / $workerCount) / 2); + $poolSize = (int)($instanceConnections / $workerCount); foreach ($connections as $key => $connection) { $type = $connection['type'] ?? ''; @@ -245,6 +238,7 @@ $buildPoolGroup = function (string $configPrefix, callable $adapterFactory, ?int $dsn = $dsn[1] ?? ''; $config[] = $name; if (empty($dsn)) { + //throw new Exception(Exception::GENERAL_SERVER_ERROR, "Missing value for DSN connection in {$key}"); continue; } @@ -260,6 +254,13 @@ $buildPoolGroup = function (string $configPrefix, callable $adapterFactory, ?int throw new Exception(Exception::GENERAL_SERVER_ERROR, "Invalid console database scheme"); } + /** + * Get Resource + * + * Creation could be reused across connection types like database, cache, queue, etc. + * + * Resource assignment to an adapter will happen below. + */ $resource = match ($dsnScheme) { 'mysql', 'mariadb' => function () use ($dsnHost, $dsnPort, $dsnUser, $dsnPass, $dsnDatabase) { @@ -286,8 +287,10 @@ $buildPoolGroup = function (string $configPrefix, callable $adapterFactory, ?int default => throw new Exception(Exception::GENERAL_SERVER_ERROR, 'Invalid scheme'), }; - $poolAdapter = $adapterFactory(); + $poolAdapter = System::getEnv('COROUTINE_POOLS', 'disabled') === 'enabled' ? new SwoolePool() : new StackPool(); + $pool = new Pool($poolAdapter, $name, $poolSize, function () use ($type, $resource, $dsn) { + // Get Adapter switch ($type) { case 'database': $adapter = match ($dsn->getScheme()) { @@ -295,6 +298,7 @@ $buildPoolGroup = function (string $configPrefix, callable $adapterFactory, ?int 'mysql' => new MySQL($resource()), default => null }; + $adapter->setDatabase($dsn->getPath()); return $adapter; case 'pubsub': @@ -318,25 +322,14 @@ $buildPoolGroup = function (string $configPrefix, callable $adapterFactory, ?int } }); - if ($syncTimeout !== null) { - $pool->setSynchronizationTimeout($syncTimeout); - } - $group->add($pool); } - Config::setParam($configPrefix . '-' . $key, $config); + Config::setParam('pools-' . $key, $config); } return $group; -}; - -$register->set('pools', fn () => $buildPoolGroup('pools', fn () => new Stack(), null)); - -/** - * Separate pool group for async/realtime contexts, using Swoole adapter and 10s sync timeout. - */ -$register->set('coroutinepools', fn () => $buildPoolGroup('coroutinepools', fn () => new SwoolePool(), 10)); +}); $register->set('db', function () { // This is usually for our workers or CLI commands scope diff --git a/app/realtime.php b/app/realtime.php index d25d952eff..fab0ce7561 100644 --- a/app/realtime.php +++ b/app/realtime.php @@ -62,7 +62,7 @@ if (!function_exists('getConsoleDB')) { global $register; /** @var Group $pools */ - $pools = $register->get('coroutinepools'); + $pools = $register->get('pools'); $adapter = new DatabasePool($pools->get('console')); $database = new Database($adapter, getCache()); @@ -92,7 +92,7 @@ if (!function_exists('getProjectDB')) { global $register; /** @var Group $pools */ - $pools = $register->get('coroutinepools'); + $pools = $register->get('pools'); if ($project->isEmpty() || $project->getId() === 'console') { return getConsoleDB(); @@ -144,7 +144,7 @@ if (!function_exists('getCache')) { global $register; - $pools = $register->get('coroutinepools'); /** @var Group $pools */ + $pools = $register->get('pools'); /** @var Group $pools */ $list = Config::getParam('pools-cache', []); $adapters = []; @@ -445,7 +445,7 @@ $server->onWorkerStart(function (int $workerId) use ($server, $register, $stats, } $start = time(); - $pubsub = new PubSubPool($register->get('coroutinepools')->get('pubsub')); + $pubsub = new PubSubPool($register->get('pools')->get('pubsub')); if ($pubsub->ping(true)) { $attempts = 0; @@ -519,7 +519,7 @@ $server->onOpen(function (int $connection, SwooleRequest $request) use ($server, Console::info("Connection open (user: {$connection})"); - App::setResource('pools', fn () => $register->get('coroutinepools')); + App::setResource('pools', fn () => $register->get('pools')); App::setResource('request', fn () => $request); App::setResource('response', fn () => $response); diff --git a/docker-compose.yml b/docker-compose.yml index 3b935b84fb..ddfbcf421b 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -298,6 +298,7 @@ services: - _APP_LOGGING_CONFIG - _APP_LOGGING_CONFIG_REALTIME - _APP_DATABASE_SHARED_TABLES + - COROUTINE_POOLS=enabled appwrite-worker-audits: entrypoint: worker-audits