From e0de0fdf6f85d70620aa98312a29372f4682db47 Mon Sep 17 00:00:00 2001 From: "paseo-bot[bot]" <266920839+paseo-bot[bot]@users.noreply.github.com> Date: Fri, 16 Oct 2026 03:15:23 +0200 Subject: [PATCH] Read cached chats while other agents stream (#7074) * perf(app): overlay pending replica rows on read instead of forcing persists readRows flushed every pending change to storage, read, and retried whenever the host revision moved. Every stream reducer flush (48 ms) and every agent delta advanced that revision, so opening another agent or workspace on a host with a streaming agent re-serialized the whole pending batch per iteration and could spin until the stream paused. Reads now run on the write queue, so no drained batch is mid-apply while a read runs, and rows still pending in memory are overlaid for the requested keys. A read never forces a persist and never waits for the host to go quiet. Persist retries after a storage failure back off exponentially (1 s to 60 s) and log the failure instead of re-serializing every second. Before: one readAgent during a 200-commit stream cost 201 store reads and 200 forced persists; a permanently failing store was written 30 times in 30 s. After: 1 read, 0 persists; 5 writes in 30 s. Fixes #4813 Refs #4814 Co-Authored-By: Claude Fable 5.1 * fix(app): log replica persist retry failures --------- Co-authored-by: Ivan Malison Co-authored-by: Claude Fable 5.1 Co-authored-by: paseo-bot[bot] <266920839+paseo-bot[bot]@users.noreply.github.com> Co-authored-by: paseo-bot[bot] --- docs/data-model.md | 5 +- .../src/runtime/replica-cache/index.test.ts | 148 +++++++++++++++++- .../app/src/runtime/replica-cache/index.ts | 83 +++++++--- 3 files changed, 215 insertions(+), 21 deletions(-) diff --git a/docs/data-model.md b/docs/data-model.md index 25a214b11c..1b578bb6d8 100644 --- a/docs/data-model.md +++ b/docs/data-model.md @@ -589,7 +589,10 @@ host, kinds, and ids. Opening a cached workspace uses exact workspace and projec directory scan. One invalid row is deleted and returned as a miss without affecting other rows. An invalid directory row and its affected checkpoint cursor are repaired in one transaction, so a later launch cannot accept a checkpoint for a partial baseline. Directory changes and their -checkpoint are also applied in one transaction. +checkpoint are also applied in one transaction. Reads wait for accepted changes to reach storage, +then retry only when a requested row changes during the read. Unrelated agent streams do not +invalidate the read. Directory baseline replacements and host removal or renaming still invalidate +the affected reads. The cache is capped at 32 MiB and evicts whole hosts in least-recently-written order. Budget bookkeeping may scan opaque row sizes during a deferred write, never during host registry startup or diff --git a/packages/app/src/runtime/replica-cache/index.test.ts b/packages/app/src/runtime/replica-cache/index.test.ts index 76e9e66464..15adc841fb 100644 --- a/packages/app/src/runtime/replica-cache/index.test.ts +++ b/packages/app/src/runtime/replica-cache/index.test.ts @@ -464,6 +464,19 @@ describe("ReplicaCache", () => { expect((await cache.readDirectory(SERVER_ID)).projects.size).toBe(0); }); + it("does not return an accepted agent that storage rejected", async () => { + const storage = new MemoryStorage(); + const cache = createCache(storage); + storage.persistentWriteFailure = new Error("disk unavailable"); + commitDirectory(cache, SERVER_ID, directory()); + await cache.flush(); + + const result = await cache.readAgent(SERVER_ID, "agent-1"); + + expect(storage.rows.size).toBe(0); + expect(result).toBeUndefined(); + }); + it("fails closed when an accepted deletion cannot be persisted before a read", async () => { const storage = new MemoryStorage(); const cache = createCache(storage); @@ -504,6 +517,25 @@ describe("ReplicaCache", () => { expect((await cache.readWorkspace(SERVER_ID, "workspace-1"))?.workspace.id).toBe("workspace-1"); }); + it("discards an in-flight read across host removal and reactivation", async () => { + const storage = new MemoryStorage(); + const cache = createCache(storage); + commitDirectory(cache, SERVER_ID, directory()); + await cache.flush(); + const started = deferred(); + const release = deferred(); + storage.onRead = started.resolve; + storage.readGate = release.promise; + + const reading = cache.readAgent(SERVER_ID, "agent-1"); + await started.promise; + cache.setHosts([]); + cache.setHosts([SERVER_ID]); + release.resolve(); + + expect(await reading).toBeUndefined(); + }); + it("discards a durable read when the host changes while it is in flight", async () => { const storage = new MemoryStorage(); const cache = createCache(storage); @@ -522,6 +554,36 @@ describe("ReplicaCache", () => { expect(await reading).toBeUndefined(); }); + it("retries a timeline read when that same timeline changes in flight", async () => { + const storage = new MemoryStorage(); + const cache = createCache(storage); + cache.commitTimeline(SERVER_ID, "agent-1", timeline("Old")); + await cache.flush(); + storage.onRead = () => { + storage.onRead = null; + cache.commitTimeline(SERVER_ID, "agent-1", timeline("New")); + }; + + expect((await cache.readTimeline(SERVER_ID, "agent-1"))?.items).toEqual([timelineItem("New")]); + expect(storage.reads).toHaveLength(2); + }); + + it("discards a requested row removed by a directory baseline replacement in flight", async () => { + const storage = new MemoryStorage(); + const cache = createCache(storage); + commitDirectory(cache, SERVER_ID, directory()); + await cache.flush(); + const replacement = directory(); + replacement.agents.clear(); + storage.onRead = () => { + storage.onRead = null; + cache.replaceDirectoryBaseline(SERVER_ID, replacement); + }; + + expect(await cache.readAgent(SERVER_ID, "agent-1")).toBeUndefined(); + expect(storage.reads).toHaveLength(2); + }); + it("never reads a timeline older than an accepted deferred replacement", async () => { const storage = new MemoryStorage(); const cache = createCache(storage); @@ -796,7 +858,7 @@ describe("ReplicaCache", () => { ]); }); - it("retries a timeline read invalidated by a concurrent directory commit", async () => { + it("reads a persisted timeline across a concurrent directory commit", async () => { const storage = new MemoryStorage(); const cache = createCache(storage); cache.commitTimeline(SERVER_ID, "agent-1", timeline("Persisted timeline")); @@ -858,6 +920,90 @@ describe("ReplicaCache", () => { expect(storage.cleanups).toBe(1); }); + + describe.each(["timeline", "directory"] as const)( + "while another agent streams %s updates", + (kind) => { + async function streamingCache() { + const storage = new MemoryStorage(); + const cache = createCache(storage); + commitDirectory(cache, SERVER_ID, directory()); + cache.commitTimeline(SERVER_ID, "agent-1", timeline()); + await cache.flush(); + const readsBefore = storage.reads.length; + const writesBefore = storage.writes; + let chunks = 0; + storage.onRead = () => { + if (chunks >= 200) return; + chunks += 1; + if (kind === "timeline") { + cache.commitTimeline(SERVER_ID, "agent-2", { + ...timeline(`chunk ${chunks}`), + agentId: "agent-2", + }); + } else { + cache.commitDirectoryMutations( + SERVER_ID, + [{ type: "upsert", kind: "agent", id: "agent-2", value: agent("agent-2") }], + { agents: { generation: "g", afterSeq: 12 + chunks } }, + ); + } + }; + return { + cache, + expectReads: (count: number, writes = 0) => { + expect(storage.reads.length - readsBefore).toBe(count); + expect(storage.writes - writesBefore).toBe(writes); + expect(chunks).toBe(count); + }, + }; + } + + it("reads another cached agent without waiting for the stream to stop", async () => { + const { cache, expectReads } = await streamingCache(); + + expect((await cache.readAgent(SERVER_ID, "agent-1"))?.title).toBe("Cached agent"); + expectReads(1); + await cache.flush(); + }); + + it("reads another cached workspace without waiting for the stream to stop", async () => { + const { cache, expectReads } = await streamingCache(); + + expect((await cache.readWorkspace(SERVER_ID, "workspace-1"))?.workspace.name).toBe("main"); + expectReads(2, 1); + await cache.flush(); + }); + + it("reads another cached timeline without waiting for the stream to stop", async () => { + const { cache, expectReads } = await streamingCache(); + + expect(await cache.readTimeline(SERVER_ID, "agent-1")).toEqual(timeline()); + expectReads(1); + await cache.flush(); + }); + }, + ); + + it("reads the directory after persisting a pending baseline replacement", async () => { + const storage = new MemoryStorage(); + const cache = createCache(storage); + commitDirectory(cache, SERVER_ID, directory()); + cache.commitTimeline(SERVER_ID, "agent-1", timeline()); + await cache.flush(); + const replacement = directory({ agents: { generation: "g2", afterSeq: 1 } }); + replacement.projects.clear(); + replacement.agents.set("agent-2", agent("agent-2")); + cache.replaceDirectoryBaseline(SERVER_ID, replacement); + + const read = await cache.readDirectory(SERVER_ID); + + expect([...read.agents.keys()].sort()).toEqual(["agent-1", "agent-2"]); + expect(read.projects.size).toBe(0); + expect(read.checkpoint).toEqual({ agents: { generation: "g2", afterSeq: 1 } }); + expect(await cache.readTimeline(SERVER_ID, "agent-1")).toEqual(timeline()); + expect(storage.writes).toBe(2); + }); }); it("retains semantic agent messages across disk-cache reconstruction", async () => { diff --git a/packages/app/src/runtime/replica-cache/index.ts b/packages/app/src/runtime/replica-cache/index.ts index 3fc8458551..94937d4dc5 100644 --- a/packages/app/src/runtime/replica-cache/index.ts +++ b/packages/app/src/runtime/replica-cache/index.ts @@ -880,9 +880,24 @@ function directoryEntityForRow(row: ReplicaRow): keyof DirectoryCheckpoint | und return undefined; } +interface ReplicaRead { + serverId: string; + kinds: readonly ReplicaRowKind[]; + ids?: readonly string[]; + invalidated: boolean; +} + +function matchesReplicaRead(read: ReplicaRead, row: ReplicaRowKey): boolean { + return ( + read.serverId === row.serverId && + read.kinds.includes(row.kind) && + (!read.ids || read.ids.includes(row.id)) + ); +} + export class ReplicaCache { private readonly activeServerIds = new Set(); - private readonly hostRevisions = new Map(); + private readonly activeReads = new Set(); private readonly storedRows = new Map>(); private readonly hostBytes = new Map(); private readonly hostWriteOrder = new Map(); @@ -1004,10 +1019,22 @@ export class ReplicaCache { // 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; + const read: ReplicaRead = { serverId, kinds, ids, invalidated: false }; + this.activeReads.add(read); + try { + if (!(await this.syncPending()) && this.hasPendingHostChanges(serverId)) return []; + if (read.invalidated || this.hasPendingReadChanges(read)) continue; + const rows = await this.rowStore.read(serverId, kinds, ids); + if ( + this.activeServerIds.has(serverId) && + !read.invalidated && + !this.hasPendingReadChanges(read) + ) { + return rows; + } + } finally { + this.activeReads.delete(read); + } } return []; } catch { @@ -1092,7 +1119,6 @@ export class ReplicaCache { ): void { if (!this.activeServerIds.has(serverId)) return; if (this.invalidatedHosts.has(serverId)) return; - this.advanceHostRevision(serverId); for (const mutation of mutations) { if (mutation.type === "delete") { this.queueEntityDelete(serverId, mutation.kind, mutation.id); @@ -1118,7 +1144,7 @@ export class ReplicaCache { replaceDirectoryBaseline(serverId: string, directory: CachedDirectory): void { if (!this.activeServerIds.has(serverId)) return; - this.advanceHostRevision(serverId); + this.invalidateHostReads(serverId, ["agent", "workspace", "project", "checkpoint"]); // Replacing the directory must not discard independently accepted timeline changes. for (const [key, row] of this.pendingUpserts) { if (row.serverId === serverId && row.kind !== "timeline") this.pendingUpserts.delete(key); @@ -1150,7 +1176,6 @@ export class ReplicaCache { commitTimeline(serverId: string, agentId: string, timeline: CachedTimeline): void { if (!this.activeServerIds.has(serverId)) return; if (timeline.agentId !== agentId) throw new Error("Timeline cache key does not match payload"); - this.advanceHostRevision(serverId); this.queueUpsert({ serverId, kind: "timeline", id: agentId, value: timeline }); this.schedulePersist(); } @@ -1159,7 +1184,7 @@ export class ReplicaCache { const next = new Set(serverIds); const removed = [...this.activeServerIds].filter((serverId) => !next.has(serverId)); const added = [...next].filter((serverId) => !this.activeServerIds.has(serverId)); - for (const serverId of [...removed, ...added]) this.advanceHostRevision(serverId); + for (const serverId of [...removed, ...added]) this.invalidateHostReads(serverId); this.activeServerIds.clear(); for (const serverId of next) this.activeServerIds.add(serverId); for (const serverId of removed) { @@ -1171,8 +1196,8 @@ export class ReplicaCache { } reconcileServerId(oldServerId: string, newServerId: string): void { - this.advanceHostRevision(oldServerId); - this.advanceHostRevision(newServerId); + this.invalidateHostReads(oldServerId); + this.invalidateHostReads(newServerId); const rows = this.storedRows.get(oldServerId); if (rows) { const newRows = this.storedRows.get(newServerId) ?? new Map(); @@ -1250,12 +1275,14 @@ export class ReplicaCache { } private queueUpsert(row: StructuredReplicaUpsert): void { + this.invalidateRowReads(row); const key = pendingRowKey(row); this.pendingDeletes.delete(key); this.pendingUpserts.set(key, row); } private queueDelete(key: ReplicaRowKey): void { + this.invalidateRowReads(key); const pendingKey = pendingRowKey(key); this.pendingUpserts.delete(pendingKey); this.pendingDeletes.set(pendingKey, key); @@ -1267,14 +1294,36 @@ export class ReplicaCache { ); } - private canReadHostRevision(serverId: string, revision: number): boolean { + private hasPendingReadChanges(read: ReplicaRead): boolean { + if ( + this.pendingBaselines.has(read.serverId) && + read.kinds.some((kind) => kind !== "timeline") + ) { + return true; + } return ( - this.activeServerIds.has(serverId) && - (this.hostRevisions.get(serverId) ?? 0) === revision && - !this.hasPendingHostChanges(serverId) + [...this.pendingUpserts.values()].some((row) => matchesReplicaRead(read, row)) || + [...this.pendingDeletes.values()].some((row) => matchesReplicaRead(read, row)) ); } + private invalidateRowReads(row: ReplicaRowKey): void { + for (const read of this.activeReads) { + if (matchesReplicaRead(read, row)) read.invalidated = true; + } + } + + private invalidateHostReads(serverId: string, kinds?: readonly ReplicaRowKind[]): void { + for (const read of this.activeReads) { + if ( + read.serverId === serverId && + (!kinds || kinds.some((kind) => read.kinds.includes(kind))) + ) { + read.invalidated = true; + } + } + } + private hasPendingHostChanges(serverId: string): boolean { if (this.pendingBaselines.has(serverId)) return true; for (const row of this.pendingUpserts.values()) { @@ -1286,10 +1335,6 @@ export class ReplicaCache { return false; } - private advanceHostRevision(serverId: string): void { - this.hostRevisions.set(serverId, (this.hostRevisions.get(serverId) ?? 0) + 1); - } - private drainPendingChanges(): PendingReplicaChanges { const changes = { upserts: [...this.pendingUpserts.values()],