Stop a replica cache read from spinning when the store rejects its writes (#5290)

* Stop a replica cache read from spinning when the store rejects its writes

A replica cache read waits for accepted commits to reach the row store
before it answers, and retries until the store agrees. When the store
rejects the write deterministically, that retry never makes progress: the
read loops on flush and read with no delay, burning a core and never
returning. The read now fails closed on a rejected write, so callers fall
back to the daemon instead of the loop saturating the renderer.

* Fail a read closed only when its own host has unwritten changes

A rejected write carries whatever was pending across every host, so
giving up the read on any rejection hid stored rows from hosts that had
nothing pending. A host whose rows are already in the store stays
readable while another host's write keeps failing.

* Bound the read count in the storage fake instead of the test body

---------

Co-authored-by: paseo-bot[bot] <266920839+paseo-bot[bot]@users.noreply.github.com>

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<void> | 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<ReplicaRow, "serverId" | "kind" | "id">): string {
     return `${row.serverId}:${row.kind}:${row.id}`;
@@ -47,6 +50,11 @@ class MemoryStorage implements ReplicaRowStore {
     ids?: readonly string[],
   ): Promise<ReplicaRow[]> {
     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<void> {
     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);

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<void> {
-    await this.persist();
+    await this.syncPending();
+  }
+
+  /** Resolves to whether every pending change reached the store. */
+  private async syncPending(): Promise<boolean> {
+    const persisted = await this.persist();
     await this.writeQueue.catch(() => undefined);
+    return persisted;
   }
 
   private async flushPending(): Promise<void> {
     await this.persist();
   }
 
-  private async persist(): Promise<void> {
+  private async persist(): Promise<boolean> {
     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 {