From 0aa8d402aeababd29c48a1733f4aadd05d2c47ea Mon Sep 17 00:00:00 2001 From: ArnabChatterjee20k Date: Thu, 14 May 2026 11:36:55 +0530 Subject: [PATCH] feat: streamline presence cleanup logic and enhance stale presence deletion --- app/realtime.php | 32 ----------------- .../Platform/Modules/Presences/HTTP/XList.php | 5 +-- src/Appwrite/Platform/Workers/Deletes.php | 35 ++++++++++++------- .../Realtime/Message/Handlers/Presence.php | 4 +-- 4 files changed, 25 insertions(+), 51 deletions(-) diff --git a/app/realtime.php b/app/realtime.php index e3c1b2faf2..98e90e594d 100644 --- a/app/realtime.php +++ b/app/realtime.php @@ -567,38 +567,6 @@ $server->onWorkerStart(function (int $workerId) use ($server, $register, $stats, )); $register->get('telemetry.workerCounter')->add(1); - // Orphan sweep on pod restart -> no connections in in-memory realtime connections object. - // So db must be in sync and only one coroutine shall trigger this - if ($workerId === 0) { - go(function (): void { - $hostname = \gethostname(); - if ($hostname === false || $hostname === '') { - return; - } - - try { - $consoleDB = getConsoleDB(); - $consoleDB->getAuthorization()->skip(fn () => $consoleDB->foreach('projects', function (Document $project) use ($hostname): void { - try { - $dbForProject = getProjectDB($project); - $deletionCount = $dbForProject->deleteDocuments('presenceLogs', [ - Query::equal('hostname', [$hostname]), - Query::equal('source', ['realtime']), - ]); - triggerPresenceUsage(-$deletionCount, $project); - } catch (Throwable $th) { - Console::error("Realtime startup orphan sweep failed for project {$project->getId()}: {$th->getMessage()}"); - logError($th, 'realtimeOrphanPresenceCleanup', tags: [ - 'projectId' => $project->getId(), - ]); - } - })); - } catch (Throwable $th) { - Console::error('Realtime startup orphan sweep error: ' . $th->getMessage()); - } - }); - } - $attempts = 0; $start = time(); diff --git a/src/Appwrite/Platform/Modules/Presences/HTTP/XList.php b/src/Appwrite/Platform/Modules/Presences/HTTP/XList.php index 2f5034504f..bf7e40ad84 100644 --- a/src/Appwrite/Platform/Modules/Presences/HTTP/XList.php +++ b/src/Appwrite/Platform/Modules/Presences/HTTP/XList.php @@ -95,10 +95,7 @@ class XList extends PlatformAction // should be excluded from the user provided query as user query would be used for caching only // otherwise cache will always miss due to the datetime now - $expiryFilter = Query::or([ - Query::isNull('expiresAt'), - Query::greaterThan('expiresAt', DateTime::now()), - ]); + $expiryFilter = Query::greaterThan('expiresAt', DateTime::now()); try { if ((int)$ttl > 0) { diff --git a/src/Appwrite/Platform/Workers/Deletes.php b/src/Appwrite/Platform/Workers/Deletes.php index 72dcd6b287..807f5f3623 100644 --- a/src/Appwrite/Platform/Workers/Deletes.php +++ b/src/Appwrite/Platform/Workers/Deletes.php @@ -220,6 +220,7 @@ class Deletes extends Action $this->deleteExpiredSessions($project, $getProjectDB); $this->deleteExpiredTransactions($project, $getProjectDB); $this->deleteExpiredPresences($project, $getProjectDB, $publisherForUsage); + $this->deleteStalePresences($project, $getProjectDB, $publisherForUsage); $this->deleteOldDeployments($queueForDeletes, $project, $getProjectDB); break; case DELETE_TYPE_REPORT: @@ -1750,23 +1751,31 @@ class Deletes extends Action { $dbForProject = $getProjectDB($project); - // Drop a presence if either: - // - its expiresAt has passed (HTTP-upserted rows always set this), or - // - it was created more than 30 days ago. The upsert validator caps expiresAt at - // +30 days, so nothing legitimate should outlive that. This branch catches rows - // the expiresAt branch can't — chiefly realtime presences (expiresAt is null, - // lifetime tied to the websocket) whose disconnect-time cleanup never ran. $now = DateTime::format(new \DateTime()); + + $deleted = $dbForProject->deleteDocuments('presenceLogs', [ + Query::lessThan('expiresAt', $now), + ], onError: function (Throwable $th) { + // Swallow errors to avoid breaking the cleanup process + }); + + if ($deleted > 0) { + $usage = (new UsageContext())->addMetric(METRIC_USERS_PRESENCE, -$deleted); + $publisherForUsage->enqueue(new Usage( + project: $project, + metrics: $usage->getMetrics(), + )); + } + } + + private function deleteStalePresences(Document $project, callable $getProjectDB, UsagePublisher $publisherForUsage): void + { + $dbForProject = $getProjectDB($project); + $oldestAllowed = DateTime::format((new \DateTime())->sub(\DateInterval::createFromDateString('30 days'))); $deleted = $dbForProject->deleteDocuments('presenceLogs', [ - Query::or([ - Query::and([ - Query::isNotNull('expiresAt'), - Query::lessThan('expiresAt', $now), - ]), - Query::lessThan('$createdAt', $oldestAllowed), - ]), + Query::lessThan('$createdAt', $oldestAllowed), ], onError: function (Throwable $th) { // Swallow errors to avoid breaking the cleanup process }); diff --git a/src/Appwrite/Realtime/Message/Handlers/Presence.php b/src/Appwrite/Realtime/Message/Handlers/Presence.php index a3bdbf3305..d3bf65b0e3 100644 --- a/src/Appwrite/Realtime/Message/Handlers/Presence.php +++ b/src/Appwrite/Realtime/Message/Handlers/Presence.php @@ -9,6 +9,7 @@ use Appwrite\Realtime\Message\Dispatcher; use Appwrite\Utopia\Database\Documents\User; use Closure; use Utopia\Database\Database; +use Utopia\Database\DateTime; use Utopia\Database\Document; use Utopia\Database\Validator\Authorization; use Utopia\Database\Validator\Permissions; @@ -77,8 +78,7 @@ class Presence extends Action 'userId' => $user->getId(), 'source' => 'realtime', 'status' => $status, - // Pod identity, used by the realtime worker's startup sweep to clean up rows - // orphaned by a previous incarnation of this hostname (i.e. pod crash before onClose ran). + 'expiresAt' => DateTime::format((new \DateTime())->modify('+30 days')), 'hostname' => \gethostname() ?: null, ]; if ($metadata !== null) {