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()],