From 3a2ff9790cd7d22fcf7ccd56a6aa3fa669332664 Mon Sep 17 00:00:00 2001 From: Jake Barnby Date: Fri, 28 Mar 2025 16:42:26 +1300 Subject: [PATCH] CLI scope pool use --- app/cli.php | 29 +++--------- src/Appwrite/Platform/Tasks/ScheduleBase.php | 4 -- .../Platform/Tasks/ScheduleExecutions.php | 38 ++++++++-------- .../Platform/Tasks/ScheduleFunctions.php | 45 +++++++++---------- .../Platform/Tasks/ScheduleMessages.php | 21 +++++---- 5 files changed, 58 insertions(+), 79 deletions(-) diff --git a/app/cli.php b/app/cli.php index f080217365..d11b8f3c6b 100644 --- a/app/cli.php +++ b/app/cli.php @@ -14,6 +14,7 @@ use Utopia\Cache\Cache; use Utopia\CLI\CLI; use Utopia\CLI\Console; use Utopia\Config\Config; +use Utopia\Database\Adapter\Pool as PoolAdapter; use Utopia\Database\Database; use Utopia\Database\Document; use Utopia\Database\Validator\Authorization; @@ -63,12 +64,8 @@ CLI::setResource('dbForPlatform', function ($pools, $cache) { $attempts++; try { // Prepare database connection - $dbAdapter = $pools - ->get('console') - ->pop() - ->getResource(); - - $dbForPlatform = new Database($dbAdapter, $cache); + $adapter = new PoolAdapter($pools->get('console')); + $dbForPlatform = new Database($adapter, $cache); $dbForPlatform ->setNamespace('_console') @@ -86,7 +83,6 @@ CLI::setResource('dbForPlatform', function ($pools, $cache) { $ready = true; } catch (\Throwable $err) { Console::warning($err->getMessage()); - $pools->get('console')->reclaim(); sleep($sleep); } } while ($attempts < $maxAttempts && !$ready); @@ -136,12 +132,8 @@ CLI::setResource('getProjectDB', function (Group $pools, Database $dbForPlatform return $database; } - $dbAdapter = $pools - ->get($dsn->getHost()) - ->pop() - ->getResource(); - - $database = new Database($dbAdapter, $cache); + $adapter = new PoolAdapter($pools->get($dsn->getHost())); + $database = new Database($adapter, $cache); $databases[$dsn->getHost()] = $database; $sharedTables = \explode(',', System::getEnv('_APP_DATABASE_SHARED_TABLES', '')); @@ -173,15 +165,8 @@ CLI::setResource('getLogsDB', function (Group $pools, Cache $cache) { return $database; } - $dbAdapter = $pools - ->get('logs') - ->pop() - ->getResource(); - - $database = new Database( - $dbAdapter, - $cache - ); + $adapter = new PoolAdapter($pools->get('logs')); + $database = new Database($adapter, $cache); $database ->setSharedTables(true) diff --git a/src/Appwrite/Platform/Tasks/ScheduleBase.php b/src/Appwrite/Platform/Tasks/ScheduleBase.php index a3c36cb96e..69ad75316b 100644 --- a/src/Appwrite/Platform/Tasks/ScheduleBase.php +++ b/src/Appwrite/Platform/Tasks/ScheduleBase.php @@ -132,8 +132,6 @@ abstract class ScheduleBase extends Action $latestDocument = \end($results); } - $pools->reclaim(); - Console::success("{$total} resources were loaded in " . (\microtime(true) - $loadStart) . " seconds"); Console::success("Starting timers at " . DateTime::now()); @@ -198,8 +196,6 @@ abstract class ScheduleBase extends Action $lastSyncUpdate = $time; $timerEnd = \microtime(true); - $pools->reclaim(); - Console::log("Sync tick: {$total} schedules were updated in " . ($timerEnd - $timerStart) . " seconds"); }); diff --git a/src/Appwrite/Platform/Tasks/ScheduleExecutions.php b/src/Appwrite/Platform/Tasks/ScheduleExecutions.php index 7cd76b480d..3539efec58 100644 --- a/src/Appwrite/Platform/Tasks/ScheduleExecutions.php +++ b/src/Appwrite/Platform/Tasks/ScheduleExecutions.php @@ -6,6 +6,7 @@ use Appwrite\Event\Func; use Swoole\Coroutine as Co; use Utopia\Database\Database; use Utopia\Pools\Group; +use Utopia\Queue\Publisher; class ScheduleExecutions extends ScheduleBase { @@ -29,9 +30,6 @@ class ScheduleExecutions extends ScheduleBase protected function enqueueResources(Group $pools, Database $dbForPlatform, callable $getProjectDB): void { - $queue = $pools->get('publisher')->pop(); - $connection = $queue->getResource(); - $queueForFunctions = new Func($connection); $intervalEnd = (new \DateTime())->modify('+' . self::ENQUEUE_TIMER . ' seconds'); foreach ($this->schedules as $schedule) { @@ -59,21 +57,25 @@ class ScheduleExecutions extends ScheduleBase $this->updateProjectAccess($schedule['project'], $dbForPlatform); - \go(function () use ($queueForFunctions, $schedule, $delay, $data) { + \go(function () use ($schedule, $delay, $data, $pools) { Co::sleep($delay); - $queueForFunctions->setType('schedule') - // Set functionId instead of function as we don't have $dbForProject - // TODO: Refactor to use function instead of functionId - ->setFunctionId($schedule['resource']['functionId']) - ->setExecution($schedule['resource']) - ->setMethod($data['method'] ?? 'POST') - ->setPath($data['path'] ?? '/') - ->setHeaders($data['headers'] ?? []) - ->setBody($data['body'] ?? '') - ->setProject($schedule['project']) - ->setUserId($data['userId'] ?? '') - ->trigger(); + $pools->get('publisher')->use(function(Publisher $publisher) use ($schedule, $data) { + $queueForFunctions = new Func($publisher); + + $queueForFunctions->setType('schedule') + // Set functionId instead of function as we don't have $dbForProject + // TODO: Refactor to use function instead of functionId + ->setFunctionId($schedule['resource']['functionId']) + ->setExecution($schedule['resource']) + ->setMethod($data['method'] ?? 'POST') + ->setPath($data['path'] ?? '/') + ->setHeaders($data['headers'] ?? []) + ->setBody($data['body'] ?? '') + ->setProject($schedule['project']) + ->setUserId($data['userId'] ?? '') + ->trigger(); + }); }); $dbForPlatform->deleteDocument( @@ -82,8 +84,6 @@ class ScheduleExecutions extends ScheduleBase ); unset($this->schedules[$schedule['$internalId']]); - } - - $queue->reclaim(); +} } } diff --git a/src/Appwrite/Platform/Tasks/ScheduleFunctions.php b/src/Appwrite/Platform/Tasks/ScheduleFunctions.php index 5b8e3027a7..f506cd3fd9 100644 --- a/src/Appwrite/Platform/Tasks/ScheduleFunctions.php +++ b/src/Appwrite/Platform/Tasks/ScheduleFunctions.php @@ -8,6 +8,7 @@ use Utopia\CLI\Console; use Utopia\Database\Database; use Utopia\Database\DateTime; use Utopia\Pools\Group; +use Utopia\Queue\Publisher; class ScheduleFunctions extends ScheduleBase { @@ -73,31 +74,29 @@ class ScheduleFunctions extends ScheduleBase \go(function () use ($delay, $scheduleKeys, $pools, $dbForPlatform) { \sleep($delay); // in seconds - $queue = $pools->get('publisher')->pop(); - $connection = $queue->getResource(); - foreach ($scheduleKeys as $scheduleKey) { - // Ensure schedule was not deleted - if (!\array_key_exists($scheduleKey, $this->schedules)) { - return; + foreach ($scheduleKeys as $scheduleKey) { + // Ensure schedule was not deleted + if (!\array_key_exists($scheduleKey, $this->schedules)) { + return; + } + + $schedule = $this->schedules[$scheduleKey]; + + $this->updateProjectAccess($schedule['project'], $dbForPlatform); + + $pools->get('publisher')->use(function(Publisher $publisher) use ($schedule) { + $queueForFunctions = new Func($publisher); + + $queueForFunctions + ->setType('schedule') + ->setFunction($schedule['resource']) + ->setMethod('POST') + ->setPath('/') + ->setProject($schedule['project']) + ->trigger(); + }); } - - $schedule = $this->schedules[$scheduleKey]; - - $this->updateProjectAccess($schedule['project'], $dbForPlatform); - - $queueForFunctions = new Func($connection); - - $queueForFunctions - ->setType('schedule') - ->setFunction($schedule['resource']) - ->setMethod('POST') - ->setPath('/') - ->setProject($schedule['project']) - ->trigger(); - } - - $queue->reclaim(); }); } diff --git a/src/Appwrite/Platform/Tasks/ScheduleMessages.php b/src/Appwrite/Platform/Tasks/ScheduleMessages.php index 201d5eab53..319e194b22 100644 --- a/src/Appwrite/Platform/Tasks/ScheduleMessages.php +++ b/src/Appwrite/Platform/Tasks/ScheduleMessages.php @@ -5,6 +5,7 @@ namespace Appwrite\Platform\Tasks; use Appwrite\Event\Messaging; use Utopia\Database\Database; use Utopia\Pools\Group; +use Utopia\Queue\Publisher; class ScheduleMessages extends ScheduleBase { @@ -41,25 +42,23 @@ class ScheduleMessages extends ScheduleBase } \go(function () use ($schedule, $pools, $dbForPlatform) { - $queue = $pools->get('publisher')->pop(); - $connection = $queue->getResource(); - $queueForMessaging = new Messaging($connection); + $pools->get('publisher')->use(function(Publisher $publisher) use ($schedule, $dbForPlatform) { + $queueForMessaging = new Messaging($publisher); - $this->updateProjectAccess($schedule['project'], $dbForPlatform); + $this->updateProjectAccess($schedule['project'], $dbForPlatform); - $queueForMessaging - ->setType(MESSAGE_SEND_TYPE_EXTERNAL) - ->setMessageId($schedule['resourceId']) - ->setProject($schedule['project']) - ->trigger(); + $queueForMessaging + ->setType(MESSAGE_SEND_TYPE_EXTERNAL) + ->setMessageId($schedule['resourceId']) + ->setProject($schedule['project']) + ->trigger(); + }); $dbForPlatform->deleteDocument( 'schedules', $schedule['$id'], ); - $queue->reclaim(); - unset($this->schedules[$schedule['$internalId']]); }); }