feat: distributed locking for platform-database writes

Adds two DI factories and wires them where coordination is needed: - distributedLock — skip on contention, void return. For idempotent fan-out where N pods doing the same write is wasteful but losing the race is correct. - distributedLockOrFail — blocking acquire (3s default) then throws GENERAL_RESOURCE_LOCKED (HTTP 409) on contention. For read-modify-write on shared mutable state where a silent skip would drop a user's change. Both factories: _APP_LOCKING_ENABLED kill switch (set 'disabled' for fail-open), fail-open on Redis-unreachable, and a lock.attempts telemetry counter sliced by outcome and target collection. Wired sites: - shared/api.php × 3 (distributedLock): keys.accessedAt + sdks, projects.accessedAt, users.accessedAt. Reduces redundant writes and cache-purge fan-out under request bursts on the same project. - Project/Services/Update.php × 1 (distributedLockOrFail): the services map toggle. Re-reads inside the lock so the baseline reflects concurrent updates. Two simultaneous toggles to different services no longer lose one of them. Lock key namespace: lock:platform:{collection}:{id}. Dep: premtsd-code/lock pinned to a specific commit as a development preview. Migration to utopia-php/lock is a follow-up once that package is published.
2026-05-26 13:51:13 +00:00 · 2026-04-27 07:54:56 +01:00
parent 1b8123bf62
commit fb0d43daf3
8 changed files with 272 additions and 24 deletions
@@ -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 <a href="/docs/email-delivery">learn more about setting up SMTP</a> in our docs.',
@@ -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.',
@@ -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')
+                        ])));
+                    });
                }
            }
        }
@@ -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']);
@@ -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": "*",
@@ -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": {
@@ -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';
@@ -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);