(fix): keep cache owner registrations in one key per collection

CHANGELOG.md

@@ -286,6 +286,11 @@ not change anything for an upgrade from 7.x.
     document's tenant, whichever tenant is selected on the writing `Database`, and leaves every other tenant's in
     place.
   - A written document's own attribute named `options` no longer names a collection to invalidate.
+  - Writes no longer leave a key behind in Redis each. The document cache and the `find()` query cache registered
+    every invalidation under a key of its own, and a Redis purge keeps a purged key, holding its generation, with no
+    expiry. On the Redis adapters a collection's invalidations now register as fields of one `#owners` key;
+    adapters that store no fields keep a key per invalidation, which their purge deletes. On Redis, keys matching
+    `*#owner:*` left by earlier builds are no longer read and can be deleted.
 - **Permissions and tenancy:**
   - Under shared tables, `upsertDocuments()` and `upsertDocument()` store their permission rows under the tenant, on
     an adapter with no earlier write and through `Adapter\Pool` alike.

src/Database/Cache/Owners.php

@@ -0,0 +1,57 @@
+<?php
+
+namespace Utopia\Database\Cache;
+
+use Utopia\Cache\Cache;
+
+/**
+ * Where an invalidation registers the token that owns a collection's cache while its write is in flight.
+ *
+ * A Redis purge keeps the purged key, holding its generation, with no expiry, so a key per token would outlive
+ * every write. Tokens are fields of one hash per collection instead. An adapter that keeps no fields lists none,
+ * and gives each token a key of its own, which its purge deletes.
+ */
+final readonly class Owners
+{
+    public function __construct(private Cache $cache)
+    {
+    }
+
+    public function register(string $key, string $token): bool
+    {
+        $owners = $this->getOwnersKey($key);
+        if ($this->cache->save($owners, $token, $token) === false) {
+            return false;
+        }
+
+        if ($this->isField($owners, $token)) {
+            return true;
+        }
+
+        return $this->cache->save($this->getOwnerKey($key, $token), $token) !== false;
+    }
+
+    public function find(string $key, string $token): Registration
+    {
+        $owners = $this->getOwnersKey($key);
+
+        return $this->isField($owners, $token)
+            ? new Registration($owners, $token)
+            : new Registration($this->getOwnerKey($key, $token));
+    }
+
+    private function isField(string $owners, string $token): bool
+    {
+        return \in_array($token, $this->cache->list($owners), true);
+    }
+
+    private function getOwnersKey(string $key): string
+    {
+        return $key.'#owners';
+    }
+
+    private function getOwnerKey(string $key, string $token): string
+    {
+        return $key.'#owner:'.$token;
+    }
+}

src/Database/Cache/QueryCache.php

@@ -169,7 +169,7 @@ public function invalidateCollection(Scope $scope, string $collection): void
      */
     public function blockCollection(string $key, string $token): void
     {
-        if ($this->cache->save($this->getOwnerKey($key, $token), $token) === false) {
+        if (! (new Owners($this->cache))->register($key, $token)) {
             throw new RuntimeException("Failed to register query cache owner for '{$key}'");
         }
 
@@ -186,14 +186,14 @@ public function blockCollection(string $key, string $token): void
      */
     public function activateCollection(string $key, string $token): void
     {
-        $ownerKey = $this->getOwnerKey($key, $token);
-        $owner = $this->cache->load($ownerKey, self::PERMANENT);
+        $registration = (new Owners($this->cache))->find($key, $token);
+        $owner = $this->cache->load($registration->key, self::PERMANENT, $registration->field);
         if ($owner !== false && $owner !== null && $owner !== $token) {
             throw new RuntimeException("Invalid query cache owner for '{$key}'");
         }
         $owned = $owner === $token;
-        if ($owned && ! $this->cache->purge($ownerKey)) {
-            $owner = $this->cache->load($ownerKey, self::PERMANENT);
+        if ($owned && ! $this->cache->purge($registration->key, $registration->field)) {
+            $owner = $this->cache->load($registration->key, self::PERMANENT, $registration->field);
             if ($owner !== false && $owner !== null) {
                 throw new RuntimeException("Failed to release query cache owner for '{$key}'");
             }
@@ -324,11 +324,6 @@ private function getFinishedKey(string $key): string
         return $key.'#finished';
     }
 
-    private function getOwnerKey(string $key, string $token): string
-    {
-        return $key.'#owner:'.$token;
-    }
-
     private function getStartedKey(string $key): string
     {
         return $key.'#started';

src/Database/Cache/Registration.php

@@ -0,0 +1,12 @@
+<?php
+
+namespace Utopia\Database\Cache;
+
+final readonly class Registration
+{
+    public function __construct(
+        public string $key,
+        public string $field = '',
+    ) {
+    }
+}

src/Database/Traits/Documents.php

@@ -11,6 +11,7 @@
 use Utopia\Console;
 use Utopia\Database\Adapter\Feature;
 use Utopia\Database\Attribute;
+use Utopia\Database\Cache\Owners;
 use Utopia\Database\Capability;
 use Utopia\Database\Change;
 use Utopia\Database\Database;
@@ -2588,14 +2589,13 @@ private function advanceDocumentCacheEpoch(string $collectionKey): bool
     private function blockDocumentCacheEpoch(string $collectionKey, string $token): bool
     {
         $epochKey = $collectionKey.'#epoch';
-        $ownerKey = $collectionKey.'#owner:'.$token;
-        if ($this->cache->save($ownerKey, $token) === false) {
+        if (! (new Owners($this->cache))->register($collectionKey, $token)) {
             $epoch = $this->cache->load($epochKey, self::TTL);
             if ($epoch === false || $epoch === null) {
                 return false;
             }
 
-            throw new RuntimeException("Failed to register document cache owner '{$ownerKey}'");
+            throw new RuntimeException("Failed to register document cache owner '{$token}' for '{$collectionKey}'");
         }
 
         $startedKey = $collectionKey.'#started';
@@ -2633,16 +2633,16 @@ protected function activateDocumentInvalidation(array $tokens): void
 
     private function activateDocumentCacheEpoch(string $collectionKey, string $token): void
     {
-        $ownerKey = $collectionKey.'#owner:'.$token;
-        $owner = $this->cache->load($ownerKey, self::TTL);
+        $registration = (new Owners($this->cache))->find($collectionKey, $token);
+        $owner = $this->cache->load($registration->key, self::TTL, $registration->field);
         if ($owner !== false && $owner !== null && $owner !== $token) {
-            throw new RuntimeException("Invalid document cache owner '{$ownerKey}'");
+            throw new RuntimeException("Invalid document cache owner '{$token}' for '{$collectionKey}'");
         }
         $owned = $owner === $token;
-        if ($owned && ! $this->cache->purge($ownerKey)) {
-            $owner = $this->cache->load($ownerKey, self::TTL);
+        if ($owned && ! $this->cache->purge($registration->key, $registration->field)) {
+            $owner = $this->cache->load($registration->key, self::TTL, $registration->field);
             if ($owner !== false && $owner !== null) {
-                throw new RuntimeException("Failed to release document cache owner '{$ownerKey}'");
+                throw new RuntimeException("Failed to release document cache owner '{$token}' for '{$collectionKey}'");
             }
             $owned = false;
         }

tests/e2e/Adapter/Scopes/GeneralTests.php

@@ -847,6 +847,55 @@ public function testNestedTransactionKeepsDocumentCacheEpochBlockedUntilOuterCom
         }
     }
 
+    public function testCacheInvalidationDoesNotAddRedisKeysPerWrite(): void
+    {
+        $database = $this->getDatabase();
+        if (! $database->getAdapter()->supports(Capability::Caching)) {
+            $this->markTestSkipped('Adapter does not use the document cache.');
+        }
+
+        $collection = 'ownerKeys'.ID::unique();
+        $database->createCollection(new Collection(id: $collection, attributes: [
+            Attribute::string(key: 'name', size: 64, required: true),
+        ], permissions: [
+            Permission::read(Role::any()),
+            Permission::create(Role::any()),
+            Permission::update(Role::any()),
+        ]));
+
+        $redis = new Redis();
+        $redis->connect('redis', 6379);
+        $cache = new Cache(new RedisAdapter($redis));
+        $original = $database->getCache();
+        $queryCache = $database->getQueryCache();
+        $destination = $database instanceof Mirror ? $database->getDestination() : null;
+        $destinationCache = $destination?->getCache();
+        $database->setCache($cache)->setQueryCache(new QueryCache($cache));
+
+        try {
+            $database->createDocument($collection, new Document(['$id' => 'doc0', 'name' => 'doc 0']));
+            $keys = $this->scanKeys($redis, $collection);
+
+            for ($index = 1; $index <= 10; $index++) {
+                $database->createDocument($collection, new Document(['$id' => 'doc'.$index, 'name' => 'doc '.$index]));
+            }
+            $database->withTransaction(function () use ($database, $collection): void {
+                $database->updateDocument($collection, 'doc1', new Document(['name' => 'updated']));
+                $database->updateDocument($collection, 'doc2', new Document(['name' => 'updated']));
+            });
+
+            $this->assertNotEmpty($keys);
+            $this->assertSame($keys, $this->scanKeys($redis, $collection), 'Redis keeps a purged key with no expiry, so a write must not leave a key of its own');
+            $this->assertCount(11, $database->find($collection, [Query::limit(100)]));
+        } finally {
+            $database->setQueryCache($queryCache)->setCache($original);
+            if ($destination !== null && $destinationCache !== null) {
+                $destination->setCache($destinationCache);
+            }
+            $database->deleteCollection($collection);
+        }
+    }
+
     /**
      * Test that withTransaction correctly resets inTransaction state
      * when a known exception (DuplicateException) is thrown after successful rollback.
@@ -1068,6 +1117,26 @@ public function testCacheReconnect(): void
         }
     }
 
+    /**
+     * @return array<string>
+     */
+    private function scanKeys(Redis $redis, string $collection): array
+    {
+        $keys = [];
+        $iterator = null;
+        do {
+            $batch = $redis->scan($iterator, '*'.\strtolower($collection).'*', 1000);
+            if (\is_array($batch)) {
+                \array_push($keys, ...$batch);
+            }
+        } while ($iterator > 0);
+
+        $keys = \array_values(\array_unique($keys));
+        \sort($keys);
+
+        return $keys;
+    }
+
     private function dropRedisConnection(Redis $redis): void
     {
         $id = $redis->rawCommand('CLIENT', 'ID');

tests/unit/Cache/QueryCacheTest.php

@@ -564,6 +564,75 @@ public function testActivationPurgeFailureStillPropagates(): void
         $this->assertNull($queryCache->getEntry(new Scope(), 'users', []));
     }
 
+    public function testInvalidationsDoNotAddKeysToACacheThatKeepsPurgedKeys(): void
+    {
+        $adapter = new RedisLeasableCache();
+        $queryCache = new QueryCache(new Cache($adapter));
+        $scope = new Scope(namespace: 'ns');
+        $queryCache->invalidateCollection($scope, 'users');
+        $keys = $adapter->keys();
+
+        for ($invalidation = 0; $invalidation < 20; $invalidation++) {
+            $queryCache->invalidateCollection($scope, 'users');
+        }
+
+        $this->assertSame($keys, $adapter->keys(), 'A purged key stays behind in Redis, so an invalidation must not leave a key of its own');
+        $this->assertNotNull($queryCache->getEntry($scope, 'users', []));
+    }
+
+    public function testOverlappingInvalidationsSucceedOnACacheWithoutFields(): void
+    {
+        $queryCache = new QueryCache(new Cache(new Memory()));
+        $key = $queryCache->getCollectionKey(new Scope(), 'users');
+
+        $queryCache->blockCollection($key, 'first');
+        $queryCache->blockCollection($key, 'second');
+        $queryCache->activateCollection($key, 'first');
+        $this->assertNull($queryCache->getEntry(new Scope(), 'users', []), 'The second writer is still in flight');
+        $queryCache->activateCollection($key, 'second');
+
+        $this->assertNull(
+            $queryCache->getEntry(new Scope(), 'users', []),
+            'A cache without generations cannot prove a new epoch fresh, so it stays blocked',
+        );
+    }
+
+    public function testActivationRejectsACorruptedOwnerRegistration(): void
+    {
+        $adapter = new RedisLeasableCache();
+        $queryCache = new QueryCache(new Cache($adapter));
+        $key = $queryCache->getCollectionKey(new Scope(), 'users');
+        $adapter->corruptFieldWrites();
+        $queryCache->blockCollection($key, 'owner');
+
+        try {
+            $queryCache->activateCollection($key, 'owner');
+            $this->fail('A corrupted owner registration was accepted');
+        } catch (\RuntimeException $error) {
+            $this->assertStringContainsString('Invalid query cache owner', $error->getMessage());
+        }
+
+        $this->assertNull($queryCache->getEntry(new Scope(), 'users', []));
+    }
+
+    public function testActivationPropagatesAnOwnerReleaseFailure(): void
+    {
+        $adapter = new RedisLeasableCache();
+        $queryCache = new QueryCache(new Cache($adapter));
+        $key = $queryCache->getCollectionKey(new Scope(), 'users');
+        $queryCache->blockCollection($key, 'owner');
+        $adapter->failFieldPurges();
+
+        try {
+            $queryCache->activateCollection($key, 'owner');
+            $this->fail('An owner release failure was not propagated');
+        } catch (\RuntimeException $error) {
+            $this->assertStringContainsString('Failed to release query cache owner', $error->getMessage());
+        }
+
+        $this->assertNull($queryCache->getEntry(new Scope(), 'users', []));
+    }
+
     public function testInvalidationPropagatesACacheWriteFailure(): void
     {
         $cache = new InvalidationCache();