From da12ab1600cbab5ec46409f80a820dbf70834f7d Mon Sep 17 00:00:00 2001 From: Jake Barnby Date: Mon, 19 Oct 2026 16:28:16 +1300 Subject: [PATCH] fix(documents): retire cached documents before a write the cache cannot confirm commits utopia-php/cache's circuit breaker now throws when it cannot confirm a purge. Single document writes already let that throw roll them back. Two paths still committed and then served the old cached documents once the cache recovered. Batch writes and purgeCachedCollection(): with the circuit open, the owner registration and the epoch read both take the breaker's fallback. blockDocumentCacheEpoch() read a missing epoch as nothing to invalidate and let the write go ahead. It now purges the unreadable epoch anyway. A real absence still returns false. An unconfirmed purge throws, so a batch write rolls back and purgeCachedCollection() reports the failure instead of returning as if it had purged. Schema changes: they purged the collection's documents only after the definition committed. A purge after the commit cannot be guaranteed: retrying or deferring it needs the same cache that just refused, and a lapsing marker written then fails the same way. So a write to a collection's definition now places the collection's tombstone inside its own transaction. If the cache refuses, the change rolls back through updateMetadata()'s existing rollbackOperation. Once the tombstone is stored, no fill made before the commit is served again. After the commit, activation publishes a fresh epoch, and a failed activation lapses into a new epoch after the writer timeout. The schema methods' own purges after the commit are therefore redundant for documents, and they now log a failure instead of throwing for a change that is stored. Query cache invalidation is unchanged. That adds a cache operation after the commit, so deleteCollection() and deleteRelationship() now check mayHaveCommitted() before rolling back. Otherwise a failure after the commit would recreate a dropped table, or reverse a relationship, whose metadata is already gone. What remains: a failure after the commit of the definition write is still thrown as a committed failure, as updateMetadata() documents. A reader that refilled the definition between the in-transaction purge and the commit can serve it for up to DOCUMENT_CACHE_RECHECK (60s). Co-Authored-By: Claude Opus 5.5 --- src/Database/Trait/Attributes.php | 6 +- src/Database/Trait/Collections.php | 6 +- src/Database/Trait/Documents.php | 30 ++- src/Database/Trait/Indexes.php | 6 +- src/Database/Trait/Relationships.php | 12 +- tests/unit/Cache/UnavailableCache.php | 123 +++++++++++ tests/unit/Documents/CacheUnavailableTest.php | 191 ++++++++++++++++++ 7 files changed, 362 insertions(+), 12 deletions(-) create mode 100644 tests/unit/Cache/UnavailableCache.php create mode 100644 tests/unit/Documents/CacheUnavailableTest.php diff --git a/src/Database/Trait/Attributes.php b/src/Database/Trait/Attributes.php index 3b27fd3ca..cf5cd6d2a 100644 --- a/src/Database/Trait/Attributes.php +++ b/src/Database/Trait/Attributes.php @@ -290,7 +290,7 @@ public function updateAttribute(string $collection, string $key, Update $update) ); if ($altering) { - $this->withRetries(fn () => $this->purgeCachedCollection($definition->getId())); + $this->purgeCommittedCollection($definition->getId()); } $this->withRetries(fn () => $this->purgeCachedDocumentInternal(self::METADATA, $definition->getId())); @@ -456,7 +456,7 @@ public function renameAttribute(string $collection, string $old, string $new): v operationDescription: "attribute rename '{$old}' to '{$new}'" ); - $this->withRetries(fn () => $this->purgeCachedCollection($definition->getId())); + $this->purgeCommittedCollection($definition->getId()); $listeners = $this->listens(Event::AttributeRename); if ($listeners !== []) { @@ -951,7 +951,7 @@ private function writeIndexes(Collection $definition, array $indexes): void private function purgeCollectionCaches(string $collection): void { - $this->withRetries(fn () => $this->purgeCachedCollection($collection)); + $this->purgeCommittedCollection($collection); $this->withRetries(fn () => $this->purgeCachedDocumentInternal(self::METADATA, $collection)); $listeners = $this->listens(Event::DocumentPurge); diff --git a/src/Database/Trait/Collections.php b/src/Database/Trait/Collections.php index 261880cb2..0054f7147 100644 --- a/src/Database/Trait/Collections.php +++ b/src/Database/Trait/Collections.php @@ -326,6 +326,10 @@ public function deleteCollection(string $collection): void try { $deleted = $this->silent(fn () => $this->deleteDocument(self::METADATA, $collection)); } catch (Throwable $error) { + if ($this->mayHaveCommitted($error)) { + throw $error; + } + if ($schemaDeleted) { try { $this->adapter->createCollection($collection, $currentAttributes, $currentIndexes); @@ -339,7 +343,7 @@ public function deleteCollection(string $collection): void ); } - $this->purgeCachedCollection($collection); + $this->purgeCommittedCollection($collection); } $listeners = $deleted ? $this->listens(Event::CollectionDelete) : []; diff --git a/src/Database/Trait/Documents.php b/src/Database/Trait/Documents.php index f78371fe4..33b472643 100644 --- a/src/Database/Trait/Documents.php +++ b/src/Database/Trait/Documents.php @@ -3522,6 +3522,28 @@ public function purgeCachedCollection(string $collection): void $this->queryCache?->invalidateCollection($this->getQueryCacheScope(), $collection); } + /** + * Invalidate a collection's cached documents after a schema change committed. The write to its definition already + * retired them before the commit, so a failure here leaves nothing stale and is not the caller's: it is logged. + */ + private function purgeCommittedCollection(string $collection): void + { + if ($collection === self::METADATA) { + $this->withRetries(fn () => $this->purgeCachedCollection($collection)); + + return; + } + + [$collectionKey] = $this->getCacheKeys($collection); + try { + $this->withRetries(fn () => $this->advanceDocumentCacheEpoch($collectionKey, $this->getDefinitionCacheKey($collection))); + } catch (Throwable $error) { + Console::warning("Warning: Failed to invalidate the cached documents of '{$collection}' after its schema change: ".$error->getMessage()); + } + + $this->withRetries(fn () => $this->queryCache?->invalidateCollection($this->getQueryCacheScope(), $collection)); + } + /** * Purge a document's cache slot, and once more after the open invalidation scope ends. * @@ -3542,6 +3564,9 @@ protected function purgeCachedDocumentInternal(string $collectionId, ?string $id $this->documentCacheDefinitions[$context][$collectionKey] = $this->getDefinitionCacheKey($collectionId); } } + if ($collectionId === self::METADATA && $id !== self::METADATA && isset($this->documentCacheMutations[$context])) { + $this->advanceDocumentCacheEpoch($this->getCacheBaseKeys($id)[0], $documentKey); + } if (isset($this->transactionWrites[$context])) { $this->transactionWrites[$context][\strtolower($documentKey)] = true; unset( @@ -3731,7 +3756,8 @@ private function advanceDocumentCacheEpoch(string $collectionKey, string $defini * Publish a tombstone before the write, then drop the definition that carried the previous * 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. + * 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. */ private function blockDocumentCacheEpoch(string $collectionKey, string $token, string $definitionKey): bool { @@ -3739,6 +3765,8 @@ private function blockDocumentCacheEpoch(string $collectionKey, string $token, s if (! $this->registerDocumentCacheOwner($collectionKey, $token)) { $epoch = $this->cache->load($epochKey, self::DOCUMENT_CACHE_PERMANENT); if ($epoch === false || $epoch === null) { + $this->cache->purge($epochKey); + return false; } } diff --git a/src/Database/Trait/Indexes.php b/src/Database/Trait/Indexes.php index 62b5b0155..48efa3568 100644 --- a/src/Database/Trait/Indexes.php +++ b/src/Database/Trait/Indexes.php @@ -126,7 +126,7 @@ public function renameIndex(string $collection, string $old, string $new): void operationDescription: "index rename '{$old}' to '{$new}'" ); - $this->withRetries(fn () => $this->purgeCachedCollection($definition->getId())); + $this->purgeCommittedCollection($definition->getId()); $listeners = $this->listens(Event::IndexRename); if ($listeners !== []) { @@ -169,7 +169,7 @@ public function deleteIndex(string $collection, string $key): void silentRollback: true ); - $this->withRetries(fn () => $this->purgeCachedCollection($definition->getId())); + $this->purgeCommittedCollection($definition->getId()); $listeners = $this->listens(Event::IndexDelete); if ($listeners !== []) { @@ -225,7 +225,7 @@ private function storeIndexes(string $collection, array $indexes): array operationDescription: "index creation '{$keys}'" ); - $this->withRetries(fn () => $this->purgeCachedCollection($definition->getId())); + $this->purgeCommittedCollection($definition->getId()); $listeners = $this->listens(Event::IndexCreate); if ($listeners !== []) { diff --git a/src/Database/Trait/Relationships.php b/src/Database/Trait/Relationships.php index 20296afaa..35ca22bc0 100644 --- a/src/Database/Trait/Relationships.php +++ b/src/Database/Trait/Relationships.php @@ -424,8 +424,8 @@ public function updateRelationship(string $collection, string $key, Relationship throw new DatabaseException("Failed to update relationship indexes for '{$key}': ".$error->getMessage(), previous: $error); } - $this->withRetries(fn () => $this->purgeCachedCollection($collectionId)); - $this->withRetries(fn () => $this->purgeCachedCollection($relatedCollectionId)); + $this->purgeCommittedCollection($collectionId); + $this->purgeCommittedCollection($relatedCollectionId); $listeners = $this->listens(Event::AttributeUpdate); if ($listeners !== []) { @@ -511,6 +511,10 @@ public function deleteRelationship(string $collection, string $key): void }); }); } catch (Throwable $error) { + if ($this->mayHaveCommitted($error)) { + throw $error; + } + $rollbacks = []; if ($shouldRollback) { $rollbacks[] = $side === RelationshipSide::Parent @@ -526,8 +530,8 @@ public function deleteRelationship(string $collection, string $key): void ); } - $this->withRetries(fn () => $this->purgeCachedCollection($collection->getId())); - $this->withRetries(fn () => $this->purgeCachedCollection($relatedCollection->getId())); + $this->purgeCommittedCollection($collection->getId()); + $this->purgeCommittedCollection($relatedCollection->getId()); $listeners = $this->listens(Event::AttributeDelete); if ($listeners !== []) { diff --git a/tests/unit/Cache/UnavailableCache.php b/tests/unit/Cache/UnavailableCache.php new file mode 100644 index 000000000..b22893690 --- /dev/null +++ b/tests/unit/Cache/UnavailableCache.php @@ -0,0 +1,123 @@ +cache = new RedisLeasableCache(); + } + + public function open(): void + { + $this->open = true; + } + + public function failWrites(): void + { + $this->failingWrites = true; + } + + public function recover(): void + { + $this->open = false; + $this->failingWrites = false; + } + + private function writable(): bool + { + return ! $this->open && ! $this->failingWrites; + } + + private function confirmPurge(): void + { + if (! $this->writable()) { + throw new RuntimeException('Cache purge could not be confirmed'); + } + } + + #[\Override] + public function load(string $key, int $ttl, string $hash = ''): mixed + { + return $this->open ? false : $this->cache->load($key, $ttl, $hash); + } + + #[\Override] + public function save(string $key, array|string $data, string $hash = '', int $ttl = 0): bool|string|array + { + return $this->writable() ? $this->cache->save($key, $data, $hash) : false; + } + + #[\Override] + public function touch(string $key, string $hash = ''): bool + { + return $this->writable() && $this->cache->touch($key, $hash); + } + + /** @return array */ + #[\Override] + public function list(string $key): array + { + return $this->open ? [] : $this->cache->list($key); + } + + #[\Override] + public function purge(string $key, string $hash = ''): bool + { + $this->confirmPurge(); + + return $this->cache->purge($key, $hash); + } + + #[\Override] + public function flush(): bool + { + return $this->writable() && $this->cache->flush(); + } + + #[\Override] + public function ping(): bool + { + return ! $this->open; + } + + #[\Override] + public function getSize(): int + { + return $this->open ? 0 : $this->cache->getSize(); + } + + #[\Override] + public function getName(?string $key = null): string + { + return 'unavailable'; + } + + #[\Override] + public function getGeneration(string $key): string + { + return $this->open ? '0' : $this->cache->getGeneration($key); + } + + #[\Override] + public function saveWithLease(string $key, array|string $data, string $hash, string $generation): bool|string|array + { + return $this->writable() ? $this->cache->saveWithLease($key, $data, $hash, $generation) : false; + } +} diff --git a/tests/unit/Documents/CacheUnavailableTest.php b/tests/unit/Documents/CacheUnavailableTest.php new file mode 100644 index 000000000..aff7753b2 --- /dev/null +++ b/tests/unit/Documents/CacheUnavailableTest.php @@ -0,0 +1,191 @@ + */ + public static function writes(): array + { + $writes = [ + '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'])) + ), + 'delete' => static fn (Database $database): bool => $database->deleteDocument('webhooks', 'hook'), + 'batch update' => static fn (Database $database): int => $database->updateDocuments( + 'webhooks', + new Document(['name' => 'updated']), + [Query::equal('$id', ['hook'])], + ), + ]; + + $cases = []; + foreach ([self::OPEN, self::FAILING_WRITES] as $outage) { + foreach ($writes as $name => $write) { + $cases["{$name}, {$outage}"] = [$outage, $write]; + } + } + + return $cases; + } + + /** + * @param Closure(Database): mixed $write + */ + #[DataProvider('writes')] + public function testAWriteTheCacheCannotInvalidateIsRolledBack(string $outage, Closure $write): void + { + $cache = new UnavailableCache(); + $database = $this->createDatabase($cache); + $outage === self::OPEN ? $cache->open() : $cache->failWrites(); + + $failure = $this->attempt(fn (): mixed => $write($database)); + $cache->recover(); + + $this->assertInstanceOf(Throwable::class, $failure, 'A write whose cache invalidation could not be confirmed must fail'); + + $this->assertSame('original', $this->stored($database), 'The write must be rolled back'); + $this->assertSame('original', $this->cached($database)); + } + + public function testAPurgeFailingAfterTheCommitRetiresTheCollectionInstead(): void + { + $cache = new UnavailableCache(); + $adapter = new CommitHookMemory(); + $database = $this->createDatabase($cache, $adapter); + $adapter->afterCommit = $cache->failWrites(...); + + $this->attempt(fn (): Document => $database->updateDocument('webhooks', 'hook', new Document(['name' => 'updated']))); + $cache->recover(); + + $this->assertSame('updated', $this->stored($database)); + $this->assertSame('updated', $this->cached($database), 'A committed write whose purge failed must not leave the old document cached'); + } + + public function testPurgeCachedCollectionReportsACacheThatCannotConfirmIt(): void + { + $cache = new UnavailableCache(); + $database = $this->createDatabase($cache); + $cache->open(); + + $this->expectExceptionMessage('Cache purge could not be confirmed'); + + $database->purgeCachedCollection('webhooks'); + } + + public function testASchemaChangeRetiresCachedDocumentsBeforeItCommits(): void + { + $cache = new UnavailableCache(); + $adapter = new CommitHookMemory(); + $database = $this->createDatabase($cache, $adapter); + $adapter->afterCommit = $cache->open(...); + + $this->attempt(fn () => $database->renameAttribute('webhooks', 'name', 'title')); + $cache->recover(); + + $this->assertSame(['title'], \array_map(static fn (Attribute $attribute): string => $attribute->key, $database->getCollection('webhooks')->attributes())); + $document = $database->getDocument('webhooks', 'hook'); + $this->assertSame('original', $document->getAttribute('title'), 'A document cached before a schema change must not be served once the cache recovers'); + $this->assertNull($document->getAttribute('name')); + } + + public function testASchemaChangeTheCacheCannotInvalidateIsRolledBack(): void + { + $cache = new UnavailableCache(); + $database = $this->createDatabase($cache); + $cache->open(); + + $failure = $this->attempt(fn () => $database->renameAttribute('webhooks', 'name', 'title')); + $cache->recover(); + + $this->assertInstanceOf(Throwable::class, $failure, 'A schema change whose cache invalidation could not be confirmed must fail'); + + $this->assertSame(['name'], \array_map(static fn (Attribute $attribute): string => $attribute->key, $database->getCollection('webhooks')->attributes())); + $this->assertSame('original', $this->cached($database)); + } + + /** + * @param Closure(): mixed $write + */ + private function attempt(Closure $write): ?Throwable + { + try { + $write(); + } catch (Throwable $error) { + return $error; + } + + return null; + } + + private function stored(Database $database): mixed + { + return UncachedTwin::of($database)->getDocument('webhooks', 'hook')->getAttribute('name'); + } + + private function cached(Database $database): mixed + { + return $database->getDocument('webhooks', 'hook')->getAttribute('name'); + } + + private function createDatabase(UnavailableCache $cache, DatabaseMemory $adapter = new DatabaseMemory()): Database + { + $database = new Database($adapter, new Cache($cache)); + $database + ->setDatabase('utopiaTests') + ->setNamespace('unavailable_'.\uniqid()); + $database->create(); + $database->createCollection(Collection::create(id: 'webhooks', attributes: [ + Attribute::string(key: 'name'), + ], permissions: [ + Permission::read(Role::any()), + Permission::create(Role::any()), + Permission::update(Role::any()), + Permission::delete(Role::any()), + ])); + $database->createDocument('webhooks', new Document(['$id' => 'hook', 'name' => 'original'])); + $this->assertSame('original', $this->cached($database)); + $this->assertSame('original', $this->cached($database)); + + return $database; + } +} + +final class CommitHookMemory extends DatabaseMemory +{ + public ?Closure $afterCommit = null; + + #[\Override] + public function commitTransaction(): bool + { + $committed = parent::commitTransaction(); + if ($committed && ! $this->inTransaction() && $this->afterCommit !== null) { + $afterCommit = $this->afterCommit; + $this->afterCommit = null; + $afterCommit(); + } + + return $committed; + } +}