From 25eb322ef7780f44ea0019a69055c534192a0161 Mon Sep 17 00:00:00 2001 From: Mohamed Boudra Date: Fri, 28 Aug 2026 13:18:26 +0200 Subject: [PATCH] fix(app): prevent archived workspaces returning from cache (#3975) Durable replica reads could race accepted directory replacements and restore rows that had already been removed in memory. Serialize reads behind pending writes and discard reads when the host changes in flight; local archives also suppress targeted cache restoration until removal is acknowledged. --- .../browser/workspace-archive-cache.spec.ts | 36 +++++++++ .../support/helpers/replica-cache-storage.ts | 19 ++++- .../app/e2e/support/helpers/seed-client.ts | 1 + .../directory-sync/workspace-replica.test.ts | 22 ++++++ .../directory-sync/workspace-replica.ts | 1 + .../src/runtime/replica-cache/index.test.ts | 78 +++++++++++++++++++ .../app/src/runtime/replica-cache/index.ts | 36 ++++++++- 7 files changed, 191 insertions(+), 2 deletions(-) create mode 100644 packages/app/e2e/browser/workspace-archive-cache.spec.ts diff --git a/packages/app/e2e/browser/workspace-archive-cache.spec.ts b/packages/app/e2e/browser/workspace-archive-cache.spec.ts new file mode 100644 index 0000000000..11d88b29d0 --- /dev/null +++ b/packages/app/e2e/browser/workspace-archive-cache.spec.ts @@ -0,0 +1,36 @@ +import { expect, test } from "../support/fixtures"; +import { gotoAppShell } from "../support/helpers/app"; +import { waitForWorkspaceInReplicaCache } from "../support/helpers/replica-cache-storage"; +import { seedWorkspace, type SeededWorkspace } from "../support/helpers/seed-client"; +import { + expectWorkspaceAbsentFromSidebar, + selectWorkspaceInSidebar, +} from "../support/helpers/sidebar"; +import { waitForSidebarHydration } from "../support/helpers/workspace-ui"; + +async function archiveWorkspaceOutsideTheApp(workspace: SeededWorkspace): Promise { + const result = await workspace.client.archiveWorkspace(workspace.workspaceId); + expect(result.error).toBeNull(); +} + +test.describe("Workspace archive cache coherence", () => { + test("an archived selected workspace cannot return from the durable cache", async ({ page }) => { + const workspace = await seedWorkspace({ repoPrefix: "archive-cache-" }); + + try { + await gotoAppShell(page); + await waitForSidebarHydration(page); + await selectWorkspaceInSidebar(page, workspace.workspaceId); + await waitForWorkspaceInReplicaCache(page, workspace.workspaceId); + + await archiveWorkspaceOutsideTheApp(workspace); + + await expectWorkspaceAbsentFromSidebar(page, workspace.workspaceId); + await expect(page.getByText("Workspace unavailable", { exact: true })).toBeVisible({ + timeout: 30_000, + }); + } finally { + await workspace.cleanup(); + } + }); +}); diff --git a/packages/app/e2e/support/helpers/replica-cache-storage.ts b/packages/app/e2e/support/helpers/replica-cache-storage.ts index d1957880fe..2595da2965 100644 --- a/packages/app/e2e/support/helpers/replica-cache-storage.ts +++ b/packages/app/e2e/support/helpers/replica-cache-storage.ts @@ -1,4 +1,4 @@ -import type { Page } from "@playwright/test"; +import { expect, type Page } from "@playwright/test"; const DATABASE_NAME = "paseo-replica-row-store"; const STORE_NAME = "rows"; @@ -116,6 +116,23 @@ export async function readReplicaCache(page: Page): Promise 0 ? assembleCache(rows) : null; } +export async function waitForWorkspaceInReplicaCache( + page: Page, + workspaceId: string, +): Promise { + await expect + .poll( + async () => { + const cache = await readReplicaCache(page); + return cache?.hosts?.some((host) => + host.workspaces.some((workspace) => workspace.id === workspaceId), + ); + }, + { timeout: 15_000 }, + ) + .toBe(true); +} + export async function writeReplicaCache(page: Page, value: ReplicaCacheRecord): Promise { const rows: ReplicaRowRecord[] = []; for (const host of value.hosts ?? []) { diff --git a/packages/app/e2e/support/helpers/seed-client.ts b/packages/app/e2e/support/helpers/seed-client.ts index 69129ddcb4..5abfb54c1b 100644 --- a/packages/app/e2e/support/helpers/seed-client.ts +++ b/packages/app/e2e/support/helpers/seed-client.ts @@ -69,6 +69,7 @@ export interface SeedDaemonClient { workspace: SeedWorkspaceDescriptor | null; error: string | null; }>; + archiveWorkspace(workspaceId: string): Promise<{ error: string | null }>; /** * Force the daemon to recompute its git snapshot and diff for a checkout, * mirroring the UI's manual refresh. Tests use this to make an out-of-band diff --git a/packages/app/src/runtime/directory-sync/workspace-replica.test.ts b/packages/app/src/runtime/directory-sync/workspace-replica.test.ts index cb6a9927c9..84f08b22fc 100644 --- a/packages/app/src/runtime/directory-sync/workspace-replica.test.ts +++ b/packages/app/src/runtime/directory-sync/workspace-replica.test.ts @@ -6,6 +6,10 @@ import { normalizeWorkspaceDescriptor, useSessionStore, } from "@/stores/session-store"; +import { + clearWorkspaceArchivePending, + markWorkspaceArchivePending, +} from "@/contexts/session-workspace-upserts"; import { WorkspaceDirectoryReplica } from "./workspace-replica"; function workspace(id: string, projectId = "project"): WorkspaceDescriptorPayload { @@ -208,3 +212,21 @@ it("does not invent a null-key project from a workspace update", () => { expect(session?.projects.has("fresh-project")).toBe(false); store.clearSession(serverId); }); + +it("does not restore a targeted cached workspace while its archive is pending", () => { + const serverId = "cached-workspace-during-archive"; + const workspaceId = "archived-workspace"; + const store = useSessionStore.getState(); + store.initializeSession(serverId, null as unknown as DaemonClient); + const replica = new WorkspaceDirectoryReplica(serverId); + markWorkspaceArchivePending({ serverId, workspaceId }); + + try { + replica.commitCachedWorkspace(normalizeWorkspaceDescriptor(workspace(workspaceId)), undefined); + + expect(useSessionStore.getState().sessions[serverId]?.workspaces.has(workspaceId)).toBe(false); + } finally { + clearWorkspaceArchivePending({ serverId, workspaceId }); + store.clearSession(serverId); + } +}); diff --git a/packages/app/src/runtime/directory-sync/workspace-replica.ts b/packages/app/src/runtime/directory-sync/workspace-replica.ts index 79e9fd5100..bc3a36fd56 100644 --- a/packages/app/src/runtime/directory-sync/workspace-replica.ts +++ b/packages/app/src/runtime/directory-sync/workspace-replica.ts @@ -96,6 +96,7 @@ export class WorkspaceDirectoryReplica { workspace: WorkspaceDescriptor, project: ProjectDescriptor | undefined, ): void { + if (shouldSuppressWorkspaceForLocalArchive({ serverId: this.serverId, workspace })) return; const snapshot = this.read(); snapshot.workspaces.set(workspace.id, workspace); if (project) snapshot.projects.set(project.projectId, project); diff --git a/packages/app/src/runtime/replica-cache/index.test.ts b/packages/app/src/runtime/replica-cache/index.test.ts index a7a57a6cff..252788bc2b 100644 --- a/packages/app/src/runtime/replica-cache/index.test.ts +++ b/packages/app/src/runtime/replica-cache/index.test.ts @@ -12,6 +12,14 @@ import type { ReplicaHostRows, ReplicaRow, ReplicaRowChanges, ReplicaRowStore } const SERVER_ID = "cached-host"; +function deferred(): { promise: Promise; resolve(): void } { + let resolve!: () => void; + const promise = new Promise((resolvePromise) => { + resolve = resolvePromise; + }); + return { promise, resolve }; +} + class MemoryStorage implements ReplicaRowStore { readonly rows = new Map(); readonly changes: ReplicaRowChanges[] = []; @@ -23,6 +31,8 @@ class MemoryStorage implements ReplicaRowStore { writes = 0; cleanups = 0; nextWriteFailure: Error | null = null; + readGate: Promise | null = null; + onRead: (() => void) | null = null; private key(row: Pick): string { return `${row.serverId}:${row.kind}:${row.id}`; @@ -36,6 +46,8 @@ class MemoryStorage implements ReplicaRowStore { ids?: readonly string[], ): Promise { this.reads.push({ serverId, kinds, ...(ids ? { ids } : {}) }); + this.onRead?.(); + await this.readGate; const acceptedKinds = new Set(kinds); const acceptedIds = ids ? new Set(ids) : null; return [...this.rows.values()].filter( @@ -217,6 +229,72 @@ describe("ReplicaCache", () => { expect(restoredTimeline).toEqual(timeline()); }); + it("never reads directory rows older than an accepted deferred deletion", async () => { + const storage = new MemoryStorage(); + const cache = createCache(storage); + cache.commitDirectory(SERVER_ID, directory()); + await cache.flush(); + + cache.commitDirectory(SERVER_ID, { + agents: new Map(), + workspaces: new Map(), + projects: new Map(), + }); + + expect(await cache.readAgent(SERVER_ID, "agent-1")).toBeUndefined(); + expect(await cache.readWorkspace(SERVER_ID, "workspace-1")).toBeUndefined(); + expect((await cache.readDirectory(SERVER_ID)).projects.size).toBe(0); + }); + + it("fails closed when an accepted deletion cannot be persisted before a read", async () => { + const storage = new MemoryStorage(); + const cache = createCache(storage); + cache.commitDirectory(SERVER_ID, directory()); + await cache.flush(); + storage.nextWriteFailure = new Error("disk busy"); + + cache.commitDirectory(SERVER_ID, { + agents: new Map(), + workspaces: new Map(), + projects: new Map(), + }); + + expect(await cache.readWorkspace(SERVER_ID, "workspace-1")).toBeUndefined(); + }); + + it("discards a durable read when the host changes while it is in flight", async () => { + const storage = new MemoryStorage(); + const cache = createCache(storage); + cache.commitDirectory(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.commitDirectory(SERVER_ID, { + agents: new Map(), + workspaces: new Map(), + projects: new Map(), + }); + release.resolve(); + + expect(await reading).toBeUndefined(); + }); + + it("never reads a timeline older than an accepted deferred replacement", async () => { + const storage = new MemoryStorage(); + const cache = createCache(storage); + cache.commitTimeline(SERVER_ID, "agent-1", timeline("Old")); + await cache.flush(); + + cache.commitTimeline(SERVER_ID, "agent-1", timeline("New")); + + expect((await cache.readTimeline(SERVER_ID, "agent-1"))?.items).toEqual([timelineItem("New")]); + }); + it("round-trips plugin timeline items", async () => { const storage = new MemoryStorage(); const writer = createCache(storage); diff --git a/packages/app/src/runtime/replica-cache/index.ts b/packages/app/src/runtime/replica-cache/index.ts index 9fe91025bc..3a678186d8 100644 --- a/packages/app/src/runtime/replica-cache/index.ts +++ b/packages/app/src/runtime/replica-cache/index.ts @@ -784,6 +784,7 @@ function directoryEntityForRow(row: ReplicaRow): keyof DirectoryCheckpoint | und export class ReplicaCache { private readonly activeServerIds = new Set(); + private readonly hostRevisions = new Map(); private readonly storedRows = new Map>(); private readonly hostBytes = new Map(); private readonly hostWriteOrder = new Map(); @@ -897,9 +898,13 @@ export class ReplicaCache { ids?: readonly string[], ): Promise { if (!this.activeServerIds.has(serverId)) return []; + const revision = this.hostRevisions.get(serverId) ?? 0; try { await this.prepareStore(); - return await this.rowStore.read(serverId, kinds, ids); + await this.flush(); + if (!this.canReadHostRevision(serverId, revision)) return []; + const rows = await this.rowStore.read(serverId, kinds, ids); + return this.canReadHostRevision(serverId, revision) ? rows : []; } catch { return []; } @@ -985,6 +990,7 @@ export class ReplicaCache { }, ): void { if (!this.activeServerIds.has(serverId)) return; + this.advanceHostRevision(serverId); this.clearPendingDirectoryChanges(serverId); const desiredRows: ReplicaRow[] = []; for (const agent of directory.agents.values()) { @@ -1013,6 +1019,7 @@ 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); const stored = serializeTimeline(timeline); if (stored) this.queueEntityUpsert(serverId, "timeline", agentId, stored); else this.queueEntityDelete(serverId, "timeline", agentId); @@ -1022,6 +1029,8 @@ export class ReplicaCache { setHosts(serverIds: Iterable): void { 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); this.activeServerIds.clear(); for (const serverId of next) this.activeServerIds.add(serverId); for (const serverId of removed) { @@ -1032,6 +1041,8 @@ export class ReplicaCache { } reconcileServerId(oldServerId: string, newServerId: string): void { + this.advanceHostRevision(oldServerId); + this.advanceHostRevision(newServerId); const rows = this.storedRows.get(oldServerId); if (rows) { const newRows = this.storedRows.get(newServerId) ?? new Map(); @@ -1142,6 +1153,29 @@ export class ReplicaCache { ); } + private canReadHostRevision(serverId: string, revision: number): boolean { + return ( + this.activeServerIds.has(serverId) && + (this.hostRevisions.get(serverId) ?? 0) === revision && + !this.hasPendingHostChanges(serverId) + ); + } + + private hasPendingHostChanges(serverId: string): boolean { + if (this.pendingDirectoryReplacements.has(serverId)) return true; + for (const row of this.pendingUpserts.values()) { + if (row.serverId === serverId) return true; + } + for (const row of this.pendingDeletes.values()) { + if (row.serverId === serverId) return true; + } + 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()],