Merge pull request #1043 from utopia-php/fix/retry-write-cache-purge

fix(documents): retry the cache invalidation a write waits on before refusing it

Author: abnegate

src/Database/Trait/Documents.php

@@ -3575,7 +3575,7 @@ protected function purgeCachedDocumentInternal(string $collectionId, ?string $id
             );
         }
 
-        $this->cache->purge($documentKey);
+        $this->withRetries(fn () => $this->cache->purge($documentKey));
 
         return true;
     }
@@ -3757,26 +3757,29 @@ private function advanceDocumentCacheEpoch(string $collectionKey, string $defini
      * epoch, so readers refill it with the tombstone. A write whose registration keeps failing
      * still blocks with its tombstone: its activation finishes it while the tombstone is its own,
      * and otherwise retires whatever readers filled while it ran. An epoch that cannot be read is
-     * purged all the same, so a cache that cannot confirm the purge fails the write.
+     * purged all the same, so a cache that cannot confirm the purge fails the write. Each step is
+     * retried first, so a blip shorter than the retries does not cost the write.
      */
     private function blockDocumentCacheEpoch(string $collectionKey, string $token, string $definitionKey): bool
     {
         $epochKey = $collectionKey.'#epoch';
         if (! $this->registerDocumentCacheOwner($collectionKey, $token)) {
             $epoch = $this->cache->load($epochKey, self::DOCUMENT_CACHE_PERMANENT);
             if ($epoch === false || $epoch === null) {
-                $this->cache->purge($epochKey);
+                $this->withRetries(fn () => $this->cache->purge($epochKey));
 
                 return false;
             }
         }
 
-        if ($this->cache->save($epochKey, self::DOCUMENT_CACHE_BLOCKED_PREFIX.$token.self::DOCUMENT_CACHE_SEPARATOR.\time()) === false) {
-            throw new RuntimeException("Failed to block document cache epoch '{$epochKey}'");
-        }
+        $this->withRetries(function () use ($epochKey, $token): void {
+            if ($this->cache->save($epochKey, self::DOCUMENT_CACHE_BLOCKED_PREFIX.$token.self::DOCUMENT_CACHE_SEPARATOR.\time()) === false) {
+                throw new RuntimeException("Failed to block document cache epoch '{$epochKey}'");
+            }
+        });
 
-        $this->cache->purge($collectionKey.'#started');
-        $this->purgeCachedDefinition($definitionKey);
+        $this->withRetries(fn () => $this->cache->purge($collectionKey.'#started'));
+        $this->withRetries(fn () => $this->purgeCachedDefinition($definitionKey));
 
         return true;
     }

tests/unit/Cache/UnavailableCache.php

@@ -19,6 +19,8 @@ final class UnavailableCache implements CacheAdapter, Leasable
 
     private bool $failingWrites = false;
 
+    private int $failingPurges = 0;
+
     public function __construct()
     {
         $this->cache = new RedisLeasableCache();
@@ -34,6 +36,14 @@ public function failWrites(): void
         $this->failingWrites = true;
     }
 
+    /**
+     * A blip rather than an outage: the next purges cannot be confirmed, later ones can.
+     */
+    public function failNextPurges(int $count): void
+    {
+        $this->failingPurges = $count;
+    }
+
     public function recover(): void
     {
         $this->open = false;
@@ -47,6 +57,12 @@ private function writable(): bool
 
     private function confirmPurge(): void
     {
+        if ($this->failingPurges > 0) {
+            $this->failingPurges--;
+
+            throw new RuntimeException('Cache purge could not be confirmed');
+        }
+
         if (! $this->writable()) {
             throw new RuntimeException('Cache purge could not be confirmed');
         }

tests/unit/Documents/CacheUnavailableTest.php

@@ -69,6 +69,41 @@ public function testAWriteTheCacheCannotInvalidateIsRolledBack(string $outage, C
         $this->assertSame('original', $this->cached($database));
     }
 
+    /**
+     * One Redis timeout used to refuse the write outright, surfacing to the client as a 500 for a
+     * cache that answered the very next call. A blip shorter than the retries must not cost the write.
+     *
+     * @param  Closure(Database): mixed  $write
+     */
+    #[DataProvider('blips')]
+    public function testAWriteRidesOutAPurgeBlip(Closure $write): void
+    {
+        $cache = new UnavailableCache();
+        $database = $this->createDatabase($cache);
+        $cache->failNextPurges(2);
+
+        $write($database);
+
+        $this->assertSame('updated', $this->stored($database));
+        $this->assertSame('updated', $this->cached($database), 'The document must not be served from the cache as it was before the write');
+    }
+
+    /** @return array<string, array{Closure(Database): mixed}> */
+    public static function blips(): array
+    {
+        return [
+            'update' => [static fn (Database $database): Document => $database->updateDocument('webhooks', 'hook', new Document(['name' => 'updated']))],
+            'update in a transaction' => [static fn (Database $database): mixed => $database->withTransaction(
+                static fn (): Document => $database->updateDocument('webhooks', 'hook', new Document(['name' => 'updated']))
+            )],
+            'batch update' => [static fn (Database $database): int => $database->updateDocuments(
+                'webhooks',
+                new Document(['name' => 'updated']),
+                [Query::equal('$id', ['hook'])],
+            )],
+        ];
+    }
+
     public function testAPurgeFailingAfterTheCommitRetiresTheCollectionInstead(): void
     {
         $cache = new UnavailableCache();

tests/unit/Documents/DocumentCacheInvalidationTest.php

@@ -165,16 +165,17 @@ public function testACollectionDefinitionWriteKeepsTheOtherDefinitionsCached():
         $this->assertTrue($database->getCollection('logs')->getAttribute('documentSecurity'), 'The written definition must be read again');
     }
 
+    /**
+     * A purge refused once is retried, so the cache refuses for as long as the write runs: an
+     * outage longer than the retries must still roll the write back.
+     */
     public function testAFailedPurgeInsideTheTransactionRollsTheWriteBack(): void
     {
         /** @var bool $refusing */
         $refusing = false;
         $database = $this->createDatabase(new CountingMemory(), $this->purgeRefusingCache(
             static function () use (&$refusing): bool {
-                $refused = $refusing;
-                $refusing = false;
-
-                return $refused;
+                return $refusing;
             }
         ));
         $database->createDocument('webhooks', $this->hook('hook'));
@@ -187,6 +188,7 @@ static function () use (&$refusing): bool {
         } catch (Throwable $error) {
             $failure = $error->getMessage();
         }
+        $refusing = false;
 
         $this->assertSame(self::PURGE_FAILURE, $failure, 'A failed purge inside the transaction must reach the caller');