diff --git a/app/config/errors.php b/app/config/errors.php index 07b0cd59ed..f5603f9223 100644 --- a/app/config/errors.php +++ b/app/config/errors.php @@ -54,6 +54,11 @@ return [ 'description' => 'Rate limit for the current endpoint has been exceeded. Please try again after some time.', 'code' => 429, ], + Exception::GENERAL_RESOURCE_LOCKED => [ + 'name' => Exception::GENERAL_RESOURCE_LOCKED, + 'description' => 'The requested resource is currently being modified by another request. Please retry after a brief delay.', + 'code' => 409, + ], Exception::GENERAL_SMTP_DISABLED => [ 'name' => Exception::GENERAL_SMTP_DISABLED, 'description' => 'SMTP is disabled on your Appwrite instance. You can learn more about setting up SMTP in our docs.', diff --git a/app/config/variables.php b/app/config/variables.php index c834656ff4..5dd7196266 100644 --- a/app/config/variables.php +++ b/app/config/variables.php @@ -34,6 +34,15 @@ return [ 'question' => '', 'filter' => '' ], + [ + 'name' => '_APP_LOCKING_ENABLED', + 'description' => 'Enable distributed locking for platform writes. Locks coordinate concurrent updates across API pods so that read-modify-write operations on shared documents do not lose updates. By default, set to \'enabled\'. Set to \'disabled\' as an emergency kill switch — locks become no-ops (fail-open) and concurrent writes will race.', + 'introduction' => '1.10.0', + 'default' => 'enabled', + 'required' => false, + 'question' => '', + 'filter' => '' + ], [ 'name' => '_APP_OPTIONS_FORCE_HTTPS', 'description' => 'Allows you to force HTTPS connection to your API. This feature redirects any HTTP call to HTTPS and adds the \'Strict-Transport-Security\' header to all HTTP responses. By default, set to \'enabled\'. To disable, set to \'disabled\'. This feature will work only when your ports are set to default 80 and 443, and you have set up wildcard certificates with DNS challenge.', diff --git a/app/controllers/shared/api.php b/app/controllers/shared/api.php index 7c2f527ccf..b07eec1a99 100644 --- a/app/controllers/shared/api.php +++ b/app/controllers/shared/api.php @@ -100,7 +100,8 @@ Http::init() ->inject('team') ->inject('apiKey') ->inject('authorization') - ->action(function (Http $utopia, Request $request, Database $dbForPlatform, Database $dbForProject, AuditContext $auditContext, Document $project, User $user, ?Document $session, array $servers, string $mode, Document $team, ?Key $apiKey, Authorization $authorization) { + ->inject('distributedLock') + ->action(function (Http $utopia, Request $request, Database $dbForPlatform, Database $dbForProject, AuditContext $auditContext, Document $project, User $user, ?Document $session, array $servers, string $mode, Document $team, ?Key $apiKey, Authorization $authorization, callable $distributedLock) { $route = $utopia->getRoute(); if ($route === null) { throw new AppwriteException(AppwriteException::GENERAL_ROUTE_NOT_FOUND); @@ -249,15 +250,20 @@ Http::init() } if (! $updates->isEmpty()) { - $dbForPlatform->getAuthorization()->skip(fn () => $dbForPlatform->updateDocument('keys', $dbKey->getId(), $updates)); + // Serialize concurrent per-request writes to this key across API nodes. + // Skip on contention is safe: the winner applies the same idempotent update; + // if this node observed a new SDK the other didn't, the next request catches up. + $distributedLock('lock:platform:keys:' . $dbKey->getId(), function () use ($dbForPlatform, $dbKey, $updates, $apiKey, $project, $user, $team) { + $dbForPlatform->getAuthorization()->skip(fn () => $dbForPlatform->updateDocument('keys', $dbKey->getId(), $updates)); - if (! empty($apiKey->getProjectId())) { - $dbForPlatform->getAuthorization()->skip(fn () => $dbForPlatform->purgeCachedDocument('projects', $project->getId())); - } elseif (! empty($apiKey->getUserId())) { - $dbForPlatform->getAuthorization()->skip(fn () => $dbForPlatform->purgeCachedDocument('users', $user->getId())); - } elseif (! empty($apiKey->getTeamId())) { - $dbForPlatform->getAuthorization()->skip(fn () => $dbForPlatform->purgeCachedDocument('teams', $team->getId())); - } + if (! empty($apiKey->getProjectId())) { + $dbForPlatform->getAuthorization()->skip(fn () => $dbForPlatform->purgeCachedDocument('projects', $project->getId())); + } elseif (! empty($apiKey->getUserId())) { + $dbForPlatform->getAuthorization()->skip(fn () => $dbForPlatform->purgeCachedDocument('users', $user->getId())); + } elseif (! empty($apiKey->getTeamId())) { + $dbForPlatform->getAuthorization()->skip(fn () => $dbForPlatform->purgeCachedDocument('teams', $team->getId())); + } + }); } $userClone = clone $user; @@ -388,9 +394,11 @@ Http::init() if ($project->getId() !== 'console') { $accessedAt = $project->getAttribute('accessedAt', 0); if (DateTime::formatTz(DateTime::addSeconds(new \DateTime(), -APP_PROJECT_ACCESS)) > $accessedAt) { - $authorization->skip(fn () => $dbForPlatform->updateDocument('projects', $project->getId(), new Document([ - 'accessedAt' => DateTime::now() - ]))); + $distributedLock('lock:platform:projects:' . $project->getId(), function () use ($authorization, $dbForPlatform, $project) { + $authorization->skip(fn () => $dbForPlatform->updateDocument('projects', $project->getId(), new Document([ + 'accessedAt' => DateTime::now() + ]))); + }); } } @@ -407,9 +415,11 @@ Http::init() 'accessedAt' => $user->getAttribute('accessedAt') ])); } else { - $authorization->skip(fn () => $dbForPlatform->updateDocument('users', $user->getId(), new Document([ - 'accessedAt' => $user->getAttribute('accessedAt') - ]))); + $distributedLock('lock:platform:users:' . $user->getId(), function () use ($authorization, $dbForPlatform, $user) { + $authorization->skip(fn () => $dbForPlatform->updateDocument('users', $user->getId(), new Document([ + 'accessedAt' => $user->getAttribute('accessedAt') + ]))); + }); } } } diff --git a/app/init/resources.php b/app/init/resources.php index 29506bfc9c..4c3a684d4e 100644 --- a/app/init/resources.php +++ b/app/init/resources.php @@ -8,6 +8,7 @@ use Appwrite\Event\Publisher\Migration as MigrationPublisher; use Appwrite\Event\Publisher\Screenshot as ScreenshotPublisher; use Appwrite\Event\Publisher\StatsResources as StatsResourcesPublisher; use Appwrite\Event\Publisher\Usage as UsagePublisher; +use Appwrite\Extend\Exception as AppwriteException; use Appwrite\Utopia\Database\Documents\User; use Executor\Executor; use Utopia\Abuse\Adapters\TimeLimit\Redis as TimeLimitRedis; @@ -22,6 +23,9 @@ use Utopia\Database\Document; use Utopia\Database\Validator\Authorization; use Utopia\DI\Container; use Utopia\DSN\DSN; +use Utopia\Lock\Adapter\Redis as LockRedisAdapter; +use Utopia\Lock\Exception\LockAcquireException; +use Utopia\Lock\Lock; use Utopia\Pools\Group; use Utopia\Queue\Broker\Pool as BrokerPool; use Utopia\Queue\Publisher; @@ -213,6 +217,134 @@ $container->set('timelimit', function (\Redis $redis) { }; }, ['redis']); +// Extract the collection segment ("keys" / "projects" / "users") from keys +// of the form "lock:platform:{target}:{id}" so metrics can slice by target. +// Used by both distributedLock and distributedLockOrFail factories below. +$lockTargetOf = function (string $key): string { + $parts = explode(':', $key, 4); + return $parts[2] ?? 'unknown'; +}; + +/** + * Distributed-lock factory: skip-on-contention variant. + * + * For idempotent writes where losing the race is correct (e.g., per-request + * `accessedAt` updates from N pods all writing the same value). On contention, + * the callback is silently skipped — another pod is doing the same work. + * + * Behavior: + * - Non-blocking acquire (one attempt). On conflict, skip and return. + * - Fail-open: if Redis is unreachable, run the callback unlocked + warn. + * - Kill switch: `_APP_LOCKING_ENABLED=disabled` runs the callback unlocked. + * + * Returns void — the caller can't distinguish "acquired and ran" from "skipped". + * + * Metric: `lock.attempts{outcome,target}` where outcome ∈ {acquired, skipped, + * backend_error, release_error}. + */ +$container->set('distributedLock', function (\Redis $redis, Telemetry $telemetry) use ($lockTargetOf) { + $enabled = System::getEnv('_APP_LOCKING_ENABLED', 'enabled') !== 'disabled'; + $attempts = $telemetry->createCounter('lock.attempts', null, 'Distributed lock acquire outcomes'); + + if (! $enabled) { + return function (string $key, \Closure $fn, float $ttl = 5.0): void { + $fn(); + }; + } + + return function (string $key, \Closure $fn, float $ttl = 5.0) use ($redis, $attempts, $lockTargetOf): void { + $target = $lockTargetOf($key); + $lock = new Lock(new LockRedisAdapter($redis), $key, $ttl); + + try { + $acquired = $lock->acquire(); + } catch (LockAcquireException $e) { + $attempts->add(1, ['outcome' => 'backend_error', 'target' => $target]); + Console::warning("Lock backend unavailable for {$key}, proceeding unlocked: {$e->getMessage()}"); + $fn(); + return; + } + + if (! $acquired) { + $attempts->add(1, ['outcome' => 'skipped', 'target' => $target]); + return; + } + + $attempts->add(1, ['outcome' => 'acquired', 'target' => $target]); + try { + $fn(); + } finally { + try { + $lock->release(); + } catch (\Throwable $e) { + $attempts->add(1, ['outcome' => 'release_error', 'target' => $target]); + Console::warning("Lock release failed for {$key}: {$e->getMessage()}"); + } + } + }; +}, ['redis', 'telemetry']); + +/** + * Distributed-lock factory: 409-on-contention variant. + * + * For explicit user-write endpoints where read-modify-write on shared mutable + * state must NOT silently drop a request. On contention, throws + * `Exception::GENERAL_RESOURCE_LOCKED` (HTTP 409) so the client retries. + * + * Behavior: + * - Blocking acquire with short timeout (default 3s). + * - On timeout, throws `GENERAL_RESOURCE_LOCKED`. + * - Fail-open: backend unreachable runs the callback unlocked + warning. + * - Kill switch: `_APP_LOCKING_ENABLED=disabled` runs the callback unlocked. + * - Returns the callback's return value so callers can use the result. + * + * Metric: `lock.attempts{outcome,target}` where outcome ∈ {acquired, contended, + * backend_error, release_error}. + */ +$container->set('distributedLockOrFail', function (\Redis $redis, Telemetry $telemetry) use ($lockTargetOf) { + $enabled = System::getEnv('_APP_LOCKING_ENABLED', 'enabled') !== 'disabled'; + $attempts = $telemetry->createCounter('lock.attempts', null, 'Distributed lock acquire outcomes'); + + if (! $enabled) { + return function (string $key, \Closure $fn, float $ttl = 10.0, float $waitTimeout = 3.0): mixed { + return $fn(); + }; + } + + return function (string $key, \Closure $fn, float $ttl = 10.0, float $waitTimeout = 3.0) use ($redis, $attempts, $lockTargetOf): mixed { + $target = $lockTargetOf($key); + $lock = new Lock(new LockRedisAdapter($redis), $key, $ttl); + + try { + $acquired = $lock->acquire(blocking: true, waitTimeout: $waitTimeout, retryDelay: 0.1); + } catch (LockAcquireException $e) { + $attempts->add(1, ['outcome' => 'backend_error', 'target' => $target]); + Console::warning("Lock backend unavailable for {$key}, proceeding unlocked: {$e->getMessage()}"); + return $fn(); + } + + if (! $acquired) { + $attempts->add(1, ['outcome' => 'contended', 'target' => $target]); + throw new AppwriteException( + AppwriteException::GENERAL_RESOURCE_LOCKED, + "Resource '{$key}' is currently being modified by another request. Please retry." + ); + } + + $attempts->add(1, ['outcome' => 'acquired', 'target' => $target]); + try { + return $fn(); + } finally { + try { + $lock->release(); + } catch (\Throwable $e) { + $attempts->add(1, ['outcome' => 'release_error', 'target' => $target]); + Console::warning("Lock release failed for {$key}: {$e->getMessage()}"); + } + } + }; +}, ['redis', 'telemetry']); + $container->set('deviceForLocal', function (Telemetry $telemetry) { return new Device\Telemetry($telemetry, new Local()); }, ['telemetry']); diff --git a/composer.json b/composer.json index 6312243e32..50ecd43cb3 100644 --- a/composer.json +++ b/composer.json @@ -94,8 +94,15 @@ "spomky-labs/otphp": "11.*", "webonyx/graphql-php": "15.31.*", "league/csv": "9.14.*", - "enshrined/svg-sanitize": "0.22.*" + "enshrined/svg-sanitize": "0.22.*", + "premtsd-code/lock": "dev-main#7ec10dae33b950bc0f473c51e5b165b20754760c" }, + "repositories": [ + { + "type": "vcs", + "url": "https://github.com/premtsd-code/lock" + } + ], "require-dev": { "ext-fileinfo": "*", "appwrite/sdk-generator": "*", diff --git a/composer.lock b/composer.lock index 02590020e0..99d1a4df9f 100644 --- a/composer.lock +++ b/composer.lock @@ -4,7 +4,7 @@ "Read more about it at https://getcomposer.org/doc/01-basic-usage.md#installing-dependencies", "This file is @generated automatically" ], - "content-hash": "c5ae97637fd0ec0a950044d1c33677ea", + "content-hash": "5108bf79532c9d666fb96daa9ec4b3d5", "packages": [ { "name": "adhocore/jwt", @@ -2104,6 +2104,75 @@ ], "time": "2026-04-10T01:33:53+00:00" }, + { + "name": "premtsd-code/lock", + "version": "dev-main", + "source": { + "type": "git", + "url": "https://github.com/premtsd-code/lock.git", + "reference": "7ec10dae33b950bc0f473c51e5b165b20754760c" + }, + "dist": { + "type": "zip", + "url": "https://api.github.com/repos/premtsd-code/lock/zipball/7ec10dae33b950bc0f473c51e5b165b20754760c", + "reference": "7ec10dae33b950bc0f473c51e5b165b20754760c", + "shasum": "" + }, + "require": { + "php": ">=8.1" + }, + "require-dev": { + "laravel/pint": "1.2.*", + "phpstan/phpstan": "^1.12", + "phpunit/phpunit": "^9.5" + }, + "suggest": { + "ext-redis": "Required for Utopia\\Lock\\Adapter\\Redis" + }, + "default-branch": true, + "type": "library", + "autoload": { + "psr-4": { + "Utopia\\Lock\\": "src/Lock" + } + }, + "autoload-dev": { + "psr-4": { + "Utopia\\Tests\\": "tests/Lock" + } + }, + "scripts": { + "check": [ + "./vendor/bin/phpstan analyse --level 8 src tests" + ], + "lint": [ + "./vendor/bin/pint --test" + ], + "format": [ + "./vendor/bin/pint" + ], + "test": [ + "./vendor/bin/phpunit --configuration phpunit.xml tests" + ] + }, + "license": [ + "MIT" + ], + "description": "A distributed lock library for PHP, part of the Utopia framework", + "keywords": [ + "distributed", + "lock", + "mutex", + "php", + "redis", + "utopia" + ], + "support": { + "source": "https://github.com/premtsd-code/lock/tree/main", + "issues": "https://github.com/premtsd-code/lock/issues" + }, + "time": "2026-04-26T15:50:08+00:00" + }, { "name": "psr/clock", "version": "1.0.0", @@ -8443,7 +8512,9 @@ ], "aliases": [], "minimum-stability": "dev", - "stability-flags": {}, + "stability-flags": { + "premtsd-code/lock": 20 + }, "prefer-stable": true, "prefer-lowest": false, "platform": { diff --git a/src/Appwrite/Extend/Exception.php b/src/Appwrite/Extend/Exception.php index 6fc3e88635..c47647cfe1 100644 --- a/src/Appwrite/Extend/Exception.php +++ b/src/Appwrite/Extend/Exception.php @@ -45,6 +45,7 @@ class Exception extends \Exception public const string GENERAL_SERVICE_DISABLED = 'general_service_disabled'; public const string GENERAL_UNAUTHORIZED_SCOPE = 'general_unauthorized_scope'; public const string GENERAL_RATE_LIMIT_EXCEEDED = 'general_rate_limit_exceeded'; + public const string GENERAL_RESOURCE_LOCKED = 'general_resource_locked'; public const string GENERAL_SMTP_DISABLED = 'general_smtp_disabled'; public const string GENERAL_PHONE_DISABLED = 'general_phone_disabled'; public const string GENERAL_ARGUMENT_INVALID = 'general_argument_invalid'; diff --git a/src/Appwrite/Platform/Modules/Project/Http/Project/Services/Update.php b/src/Appwrite/Platform/Modules/Project/Http/Project/Services/Update.php index 7aab6f5ad0..e7b048b734 100644 --- a/src/Appwrite/Platform/Modules/Project/Http/Project/Services/Update.php +++ b/src/Appwrite/Platform/Modules/Project/Http/Project/Services/Update.php @@ -60,6 +60,7 @@ class Update extends Action ->inject('project') ->inject('authorization') ->inject('queueForEvents') + ->inject('distributedLockOrFail') ->callback($this->action(...)); } @@ -70,14 +71,26 @@ class Update extends Action Database $dbForPlatform, Document $project, Authorization $authorization, - Event $queueForEvents + Event $queueForEvents, + callable $distributedLockOrFail, ): void { - $services = $project->getAttribute('services', []); - $services[$serviceId] = $enabled; + // The services map is a JSON object on the project document. Two + // concurrent service toggles read the same baseline, each set their + // own key, and the second updateDocument() overwrites the first — + // silent lost-update on a sparse write. Lock the project doc for + // the read-modify-write window; re-read inside the lock so the + // baseline reflects any update that landed between request init + // and lock acquisition. + $project = $distributedLockOrFail("platform:project:{$project->getId()}", function () use ($project, $serviceId, $enabled, $dbForPlatform, $authorization) { + $project = $authorization->skip(fn () => $dbForPlatform->getDocument('projects', $project->getId())); - $project = $authorization->skip(fn () => $dbForPlatform->updateDocument('projects', $project->getId(), new Document([ - 'services' => $services, - ]))); + $services = $project->getAttribute('services', []); + $services[$serviceId] = $enabled; + + return $authorization->skip(fn () => $dbForPlatform->updateDocument('projects', $project->getId(), new Document([ + 'services' => $services, + ]))); + }); $queueForEvents->setParam('serviceId', $serviceId);