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

A Redis reply that misses the 0.1s read timeout was turning stored writes
into 500s. In production every still-firing group fails after the commit:
purgeWrittenDocuments (CLOUD-3R97), the definition purge in the epoch
publish (CLOUD-3R95) and the epoch save itself (CLOUD-3R8D). Throwing
there cannot undo the write; it only reports a stored write as failed and
makes queue workers repeat it. Every step after the commit already leaves
the cache fail-closed when it cannot be confirmed, so it is now retried and
then logged instead of thrown. Purge events still fire and their listener
failures still reach the caller.

Before the commit, a document purge that outlasts its retries used to
refuse the write. It now blocks the collection's epoch instead, the marker
#1042 showed is the one that guarantees no stale read; only when that
fails too is the write refused, as during a real outage.

The finish purge is retried only while #finished is unchanged, so a purge
that timed out but landed is never counted twice.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

Author: abnegate

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'));