fix(documents): stop a cache blip from failing a write it cannot make stale

src/Database/Database.php

@@ -3323,7 +3323,7 @@ private function cleanup(
     /**
      * Persist metadata with automatic rollback on failure
      *
-     * A failure raised after the metadata write committed (its cache invalidation or events), or a commit that
+     * A failure raised after the metadata write committed (its events), or a commit that
      * could not be confirmed, is rethrown unchanged and rolls nothing back: the definition it reports on may be stored.
      *
      * @param  callable|null  $rollbackOperation  Cleanup operation to run if persistence fails (null if no cleanup needed)

src/Database/Trait/Documents.php

@@ -16,6 +16,7 @@
 use Utopia\Database\Attribute;
 use Utopia\Database\Cache\Epoch;
 use Utopia\Database\Cache\Owners;
+use Utopia\Database\Cache\Registration;
 use Utopia\Database\Cache\Scope;
 use Utopia\Database\Capability;
 use Utopia\Database\Change;
@@ -3545,7 +3546,9 @@ private function purgeCommittedCollection(string $collection): void
     }
 
     /**
-     * Purge a document's cache slot, and once more after the open invalidation scope ends.
+     * Purge a document's cache slot, and once more after the open invalidation scope ends. A write whose purge
+     * cannot be confirmed blocks its collection's epoch instead, before it commits, so the document cached
+     * before it is never served again; only when that fails too is the write refused.
      *
      * @throws Exception
      */
@@ -3575,7 +3578,20 @@ protected function purgeCachedDocumentInternal(string $collectionId, ?string $id
             );
         }
 
-        $this->withRetries(fn () => $this->cache->purge($documentKey));
+        $scoped = $collectionId !== self::METADATA && isset($this->documentCacheMutations[$context]);
+        try {
+            if ($scoped && isset($this->documentCacheMutations[$context][$collectionKey])) {
+                $this->cache->purge($documentKey);
+            } else {
+                $this->withRetries(fn () => $this->cache->purge($documentKey));
+            }
+        } catch (Throwable $error) {
+            if (! $scoped) {
+                throw $error;
+            }
+
+            $this->advanceDocumentCacheEpoch($collectionKey, $this->getDefinitionCacheKey($collectionId));
+        }
 
         return true;
     }
@@ -3619,7 +3635,7 @@ private function purgeCachedDefinitions(): void
      * Purge the documents a transaction wrote once it has committed or rolled back. A reader
      * outside the transaction may have cached the pre-commit row after the purge inside it, so
      * when this purge fails the collection's epoch is retired instead, which no such fill survives.
-     * A collection definition has no epoch of its own: its purge is tried once more.
+     * A collection definition has no epoch of its own, so its failed purge is the one thrown.
      *
      * @param  array<string, string>  $documents  Collection keys by document key
      */
@@ -3628,20 +3644,29 @@ protected function purgeWrittenDocuments(array $documents): void
         $definitions = $this->documentCacheDefinitions[$this->getEventContext()] ?? [];
         $failure = null;
         $retired = [];
+        $reachable = true;
         foreach ($documents as $documentKey => $collectionKey) {
+            if (isset($retired[$collectionKey])) {
+                continue;
+            }
             try {
-                $this->cache->purge($documentKey);
+                if ($reachable) {
+                    $this->withRetries(fn () => $this->cache->purge($documentKey));
+                } else {
+                    $this->cache->purge($documentKey);
+                }
             } catch (Throwable $error) {
-                $failure ??= $error;
+                $reachable = false;
+                if (! isset($definitions[$collectionKey])) {
+                    $failure ??= $error;
+
+                    continue;
+                }
                 try {
-                    if (! isset($definitions[$collectionKey])) {
-                        $this->cache->purge($documentKey);
-                    } elseif (! isset($retired[$collectionKey])) {
-                        $retired[$collectionKey] = true;
-                        $this->advanceDocumentCacheEpoch($collectionKey, $definitions[$collectionKey]);
-                    }
+                    $this->advanceDocumentCacheEpoch($collectionKey, $definitions[$collectionKey]);
+                    $retired[$collectionKey] = true;
                 } catch (Throwable) {
-                    // The purge failure below reaches the caller either way.
+                    $failure ??= $error;
                 }
             }
         }
@@ -3838,14 +3863,7 @@ private function activateDocumentCacheEpoch(string $collectionKey, string $token
         if ($owner !== false && $owner !== null && $owner !== $token) {
             throw new RuntimeException("Invalid document cache owner '{$token}' for '{$collectionKey}'");
         }
-        $owned = $owner === $token;
-        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 '{$token}' for '{$collectionKey}'");
-            }
-            $owned = false;
-        }
+        $owned = $owner === $token && $this->releaseDocumentCacheOwner($registration, $collectionKey, $token);
 
         $startedKey = $collectionKey.'#started';
         $finishedKey = $collectionKey.'#finished';
@@ -3874,19 +3892,7 @@ private function activateDocumentCacheEpoch(string $collectionKey, string $token
             return;
         }
 
-        $this->cache->purge($finishedKey);
-        $nextFinished = $this->cache->getGeneration($finishedKey);
-
-        // A cache flush restarts generations, so an unchanged #finished
-        // only proves this purge was lost while the epoch read with it is
-        // still in place; publish the new epoch after this check, not before.
-        if (
-            $nextFinished === $finished
-            && \is_string($epoch)
-            && $this->cache->load($epochKey, self::DOCUMENT_CACHE_PERMANENT) === $epoch
-        ) {
-            throw new RuntimeException("Failed to finish document cache invalidation '{$epochKey}'");
-        }
+        $nextFinished = $this->finishDocumentCacheWrite($collectionKey, $finished, $epoch);
 
         $nextStarted = $this->cache->getGeneration($startedKey);
         if ($nextStarted === $nextFinished) {
@@ -3912,11 +3918,68 @@ private function publishDocumentCacheEpoch(string $collectionKey, string $starte
     {
         $epochKey = $collectionKey.'#epoch';
         $epoch = self::DOCUMENT_CACHE_ACTIVE_PREFIX.\bin2hex(\random_bytes(16)).self::DOCUMENT_CACHE_SEPARATOR.$started;
-        if ($this->cache->save($epochKey, $epoch) === false) {
-            throw new RuntimeException("Failed to activate document cache epoch '{$epochKey}'");
-        }
+        $this->withRetries(function () use ($epochKey, $epoch): void {
+            if ($this->cache->save($epochKey, $epoch) === false) {
+                throw new RuntimeException("Failed to activate document cache epoch '{$epochKey}'");
+            }
+        });
 
-        $this->purgeCachedDefinition($definitionKey);
+        $this->withRetries(fn () => $this->purgeCachedDefinition($definitionKey));
+    }
+
+    /**
+     * Release this write's registration, and report whether it was still this write's to release. A purge
+     * that cannot be confirmed is retried while the registration is still in place.
+     */
+    private function releaseDocumentCacheOwner(Registration $registration, string $collectionKey, string $token): bool
+    {
+        $released = true;
+        $this->withRetries(function () use ($registration, $collectionKey, $token, &$released): void {
+            if ($this->cache->purge($registration->key, $registration->field)) {
+                return;
+            }
+
+            $owner = $this->cache->load($registration->key, self::TTL, $registration->field);
+            if ($owner !== false && $owner !== null) {
+                throw new RuntimeException("Failed to release document cache owner '{$token}' for '{$collectionKey}'");
+            }
+            $released = false;
+        });
+
+        return $released;
+    }
+
+    /**
+     * Count this write as finished and return the `#finished` generation after it. A cache flush restarts
+     * generations, so an unchanged `#finished` only proves the purge was lost while the epoch read with it
+     * is still in place; the purge is retried only then, so a purge that did land is never counted twice.
+     */
+    private function finishDocumentCacheWrite(string $collectionKey, string $finished, mixed $epoch): string
+    {
+        $finishedKey = $collectionKey.'#finished';
+        $epochKey = $collectionKey.'#epoch';
+        $next = $finished;
+        $attempted = false;
+        $this->withRetries(function () use ($finishedKey, $epochKey, $finished, $epoch, &$next, &$attempted): void {
+            if ($attempted) {
+                $next = $this->cache->getGeneration($finishedKey);
+            }
+            if ($next === $finished) {
+                $attempted = true;
+                $this->cache->purge($finishedKey);
+            }
+
+            $next = $this->cache->getGeneration($finishedKey);
+            if (
+                $next === $finished
+                && \is_string($epoch)
+                && $this->cache->load($epochKey, self::DOCUMENT_CACHE_PERMANENT) === $epoch
+            ) {
+                throw new RuntimeException("Failed to finish document cache invalidation '{$epochKey}'");
+            }
+        });
+
+        return $next;
     }
 
     /**

src/Database/Trait/Transactions.php

@@ -294,26 +294,9 @@ private function withInvalidationScope(callable $callback): mixed
                 $this->documentPurgeEvents[$context],
             );
 
-            $failure = null;
-            try {
-                $this->purgeWrittenDocuments($documents);
-            } catch (Throwable $error) {
-                $failure = $error;
-            }
-            try {
-                $this->activateDocumentInvalidation($documentTokens);
-            } catch (Throwable $error) {
-                $failure ??= $error;
-            }
-            try {
-                $this->activateInvalidation($queryTokens);
-            } catch (Throwable $error) {
-                $failure ??= $error;
-            }
-
-            $announcement = $this->announceDocumentPurges($purgeEvents);
-            $failure ??= $announcement;
+            $this->invalidateCommitted($documents, $documentTokens, $queryTokens);
 
+            $failure = $this->announceDocumentPurges($purgeEvents);
             if ($failure !== null) {
                 $this->committedFailures ??= new WeakMap();
                 $this->committedFailures[$failure] = true;
@@ -327,6 +310,32 @@ private function withInvalidationScope(callable $callback): mixed
         return $result;
     }
 
+    /**
+     * Invalidate the cache for writes that have committed. The writes are stored whatever the cache answers, and each
+     * step that cannot be confirmed leaves the cache fail-closed: a tombstone left in place lapses into a new epoch,
+     * never back to the old one. A failure is logged, as throwing it would report a stored write as failed.
+     *
+     * @param  array<string, string>  $documents  Collection keys by document key
+     * @param  array<string, string>  $documentTokens
+     * @param  array<string, string>  $queryTokens
+     */
+    private function invalidateCommitted(array $documents, array $documentTokens, array $queryTokens): void
+    {
+        $steps = [
+            fn () => $this->purgeWrittenDocuments($documents),
+            fn () => $this->activateDocumentInvalidation($documentTokens),
+            fn () => $this->activateInvalidation($queryTokens),
+        ];
+
+        foreach ($steps as $step) {
+            try {
+                $step();
+            } catch (Throwable $error) {
+                Console::warning('Warning: Failed to invalidate the cache after a commit: '.$error->getMessage());
+            }
+        }
+    }
+
     private function queueDefinitionRefill(string $definitionKey, string $field, string $id): void
     {
         $tenant = $this->adapter->getTenant();
@@ -370,7 +379,7 @@ private function cacheTransactionDefinitions(array $refills): void
 
     /**
      * Whether the error was raised after its outermost transaction committed: the writes it
-     * reports on are stored, and only the invalidation or the events after the commit failed.
+     * reports on are stored, and only the events after the commit failed.
      */
     private function failedAfterCommit(Throwable $error): bool
     {

tests/unit/Cache/DatabaseTest.php

@@ -616,12 +616,8 @@ public function testActivationFailureAfterCommitLeavesSharedEpochBlocked(): void
             );
             $cache->failActivations();
 
-            try {
-                $writer->updateDocument('users', 'existing', new Document(['name' => 'updated']));
-                $this->fail('Post-commit query-cache activation failure was not propagated');
-            } catch (\RuntimeException $exception) {
-                $this->assertStringContainsString('activate query cache', $exception->getMessage());
-            }
+            $updated = $writer->updateDocument('users', 'existing', new Document(['name' => 'updated']));
+            $this->assertSame('updated', $updated->getAttribute('name'), 'A stored write must not be reported as failed because its query-cache activation failed');
 
             $readerAdapter->observeFinds('users');
             $this->assertSame(

tests/unit/Cache/UnavailableCache.php

@@ -21,6 +21,8 @@ final class UnavailableCache implements CacheAdapter, Leasable
 
     private int $failingPurges = 0;
 
+    private int $failingSaves = 0;
+
     public function __construct()
     {
         $this->cache = new RedisLeasableCache();
@@ -44,6 +46,14 @@ public function failNextPurges(int $count): void
         $this->failingPurges = $count;
     }
 
+    /**
+     * A blip on writes: the next saves report false, later ones are stored.
+     */
+    public function failNextSaves(int $count): void
+    {
+        $this->failingSaves = $count;
+    }
+
     public function recover(): void
     {
         $this->open = false;
@@ -77,6 +87,12 @@ public function load(string $key, int $ttl, string $hash = ''): mixed
     #[\Override]
     public function save(string $key, array|string $data, string $hash = '', int $ttl = 0): bool|string|array
     {
+        if ($this->failingSaves > 0) {
+            $this->failingSaves--;
+
+            return false;
+        }
+
         return $this->writable() ? $this->cache->save($key, $data, $hash) : false;
     }
 

tests/unit/CoreMinorsTest.php

@@ -337,7 +337,7 @@ public function testCreateIndexKeepsItsIndexWhenTheInvalidationAfterTheCommitFai
 
         $this->assertSame(0, $deletes, 'An index whose definition committed must not be rolled back');
         $this->assertSame(1, $writes, 'A write that committed must not be repeated');
-        $this->assertSame($failures[0] ?? null, $error, 'The failure after the commit must reach the caller as it was raised');
+        $this->assertNull($error, 'A stored write must not be reported as failed because the invalidation after its commit failed');
         $this->assertSame(['by_count'], $this->indexKeys($database, 'logs'));
         $this->assertTrue($this->hasSchemaIndex($database, 'logs', 'by_count'), 'The committed index must still exist');
     }
@@ -423,7 +423,7 @@ public function testCreateRelationshipKeepsItsWorkWhenTheInvalidationAfterTheCom
         $this->assertSame(['_index_profile'], $this->indexKeys($database, 'accounts'), 'The two-way index must still be created');
         $this->assertTrue($this->hasSchemaIndex($database, 'profiles', '_index_account'));
         $this->assertTrue($this->hasSchemaIndex($database, 'accounts', '_index_profile'));
-        $this->assertSame($failures[0] ?? null, $error, 'The failure after the commit must reach the caller as it was raised');
+        $this->assertNull($error, 'A stored write must not be reported as failed because the invalidation after its commit failed');
     }
 
     /**

tests/unit/CreateCollectionCleanupTest.php

@@ -98,10 +98,9 @@ public function commitTransaction(): bool
         $adapter->afterCommit = null;
         $cache->failing = false;
 
-        $this->assertNotNull($error, 'A failed invalidation after the commit must reach the caller');
+        $this->assertNull($error, 'A stored collection must not be reported as failed because the invalidation after its commit failed');
         $this->assertTrue($adapter->collectionExists('cleanup', 'logs'), 'A collection whose definition committed must keep its table');
         $this->assertSame('logs', $database->getCollection('logs')->getId());
-        $this->assertSame($failure, $error, 'The failure after the commit must reach the caller as it was raised');
 
         $database->createDocument('logs', new Document(['message' => 'kept']));
         $this->assertSame(1, $database->count('logs'));