diff --git a/packages/app/src/runtime/replica-cache/index.test.ts b/packages/app/src/runtime/replica-cache/index.test.ts index e4e759bc15..6c273a325c 100644 --- a/packages/app/src/runtime/replica-cache/index.test.ts +++ b/packages/app/src/runtime/replica-cache/index.test.ts @@ -32,8 +32,11 @@ class MemoryStorage implements ReplicaRowStore { writes = 0; cleanups = 0; nextWriteFailure: Error | null = null; + persistentWriteFailure: Error | null = null; readGate: Promise | null = null; onRead: (() => void) | null = null; + /** Throws once a read goes past this count, so a non-terminating read loop fails the test. */ + readLimit: number | null = null; private key(row: Pick): string { return `${row.serverId}:${row.kind}:${row.id}`; @@ -47,6 +50,11 @@ class MemoryStorage implements ReplicaRowStore { ids?: readonly string[], ): Promise { this.reads.push({ serverId, kinds, ...(ids ? { ids } : {}) }); + if (this.readLimit !== null && this.reads.length > this.readLimit) { + throw new Error( + `replica cache read ${this.reads.length} passed the limit of ${this.readLimit}`, + ); + } this.onRead?.(); await this.readGate; const acceptedKinds = new Set(kinds); @@ -71,6 +79,7 @@ class MemoryStorage implements ReplicaRowStore { async apply(changes: ReplicaRowChanges): Promise { this.writes += 1; + if (this.persistentWriteFailure) throw this.persistentWriteFailure; if (this.nextWriteFailure) { const error = this.nextWriteFailure; this.nextWriteFailure = null; @@ -432,6 +441,34 @@ describe("ReplicaCache", () => { expect(await cache.readWorkspace(SERVER_ID, "workspace-1")).toBeUndefined(); }); + it("gives up a read instead of retrying while the store keeps rejecting writes", async () => { + const storage = new MemoryStorage(); + const cache = createCache(storage); + commitDirectory(cache, SERVER_ID, directory()); + await cache.flush(); + storage.persistentWriteFailure = new Error("QuotaExceededError"); + storage.reads.length = 0; + storage.readLimit = 5; + + deleteDirectory(cache, SERVER_ID); + + expect(await cache.readWorkspace(SERVER_ID, "workspace-1")).toBeUndefined(); + expect(storage.reads.length).toBeLessThanOrEqual(1); + }); + + it("still reads a host whose rows are stored when another host's write is rejected", async () => { + const storage = new MemoryStorage(); + const cache = new ReplicaCache(storage, noLegacyCleanup); + cache.setHosts([SERVER_ID, "other-host"]); + commitDirectory(cache, SERVER_ID, directory()); + await cache.flush(); + storage.persistentWriteFailure = new Error("QuotaExceededError"); + + commitDirectory(cache, "other-host", directory()); + + expect((await cache.readWorkspace(SERVER_ID, "workspace-1"))?.workspace.id).toBe("workspace-1"); + }); + it("discards a durable read when the host changes while it is in flight", async () => { const storage = new MemoryStorage(); const cache = createCache(storage); diff --git a/packages/app/src/runtime/replica-cache/index.ts b/packages/app/src/runtime/replica-cache/index.ts index 4a6e284b32..bce45d3b51 100644 --- a/packages/app/src/runtime/replica-cache/index.ts +++ b/packages/app/src/runtime/replica-cache/index.ts @@ -983,7 +983,11 @@ export class ReplicaCache { try { await this.prepareStore(); while (this.activeServerIds.has(serverId)) { - await this.flush(); + // A read may only answer from rows the store already holds, so it waits for this host's + // accepted commits to land. A store that rejects them will keep rejecting them: fail closed + // rather than re-attempt the write on every pass. A write rejected for another host leaves + // this host's stored rows readable. + if (!(await this.syncPending()) && this.hasPendingHostChanges(serverId)) return []; const revision = this.hostRevisions.get(serverId) ?? 0; const rows = await this.rowStore.read(serverId, kinds, ids); if (this.canReadHostRevision(serverId, revision)) return rows; @@ -1173,20 +1177,26 @@ export class ReplicaCache { } async flush(): Promise { - await this.persist(); + await this.syncPending(); + } + + /** Resolves to whether every pending change reached the store. */ + private async syncPending(): Promise { + const persisted = await this.persist(); await this.writeQueue.catch(() => undefined); + return persisted; } private async flushPending(): Promise { await this.persist(); } - private async persist(): Promise { + private async persist(): Promise { if (this.persistTimer) { clearTimeout(this.persistTimer); this.persistTimer = null; } - if (!this.hasPendingChanges()) return; + if (!this.hasPendingChanges()) return true; const write = this.writeQueue .catch(() => undefined) .then(async () => { @@ -1206,11 +1216,16 @@ export class ReplicaCache { } catch { this.restorePendingChanges(pending); if (this.hasPendingChanges()) this.schedulePersist(); + return false; } - return undefined; + return true; }); - this.writeQueue = write; - await write; + // The queue only sequences writes; every consumer decides for itself what a failure means. + this.writeQueue = write.then( + () => undefined, + () => undefined, + ); + return write; } private queueEntityDelete(serverId: string, kind: ReplicaRowKind, id: string): void {