From d2e2154a8a039261369f92973aac9c1e43b28683 Mon Sep 17 00:00:00 2001 From: Mohamed Boudra Date: Sat, 26 Sep 2026 16:58:53 +0200 Subject: [PATCH] fix(app): keep restored workspaces during directory refresh (#5394) * fix(app): keep restored workspaces during directory refresh Capture workspace maps and cursors from one transaction baseline so cache hydration cannot pair an empty map with a current cursor. Reject legacy directory checkpoints to recover caches already persisted with missing workspaces. * test(app): clarify workspace cache recovery flow --- .../app/e2e/browser/sidebar-workspace.spec.ts | 202 +++++++++++++ .../src/runtime/directory-sync/index.test.ts | 282 +++++++++++++++++- .../app/src/runtime/directory-sync/index.ts | 44 ++- .../src/runtime/replica-cache/index.test.ts | 35 +++ .../app/src/runtime/replica-cache/index.ts | 25 +- 5 files changed, 567 insertions(+), 21 deletions(-) diff --git a/packages/app/e2e/browser/sidebar-workspace.spec.ts b/packages/app/e2e/browser/sidebar-workspace.spec.ts index 85b52c1581..b588add807 100644 --- a/packages/app/e2e/browser/sidebar-workspace.spec.ts +++ b/packages/app/e2e/browser/sidebar-workspace.spec.ts @@ -1,4 +1,5 @@ import path from "node:path"; +import type { Page, TestInfo } from "@playwright/test"; import { test, expect } from "../support/fixtures"; import { gotoAppShell } from "../support/helpers/app"; import { @@ -11,6 +12,7 @@ import { import { seedWorkspace } from "../support/helpers/seed-client"; import { expectWorkspaceHeader } from "../support/helpers/workspace-ui"; import { getServerId } from "../support/helpers/server-id"; +import { getE2EDaemonPort } from "../support/helpers/daemon-port"; import { projectEquivalenceViewKey } from "../support/helpers/project-view-key"; import { escapeRegex } from "../support/helpers/regex"; import { openFilesPanel } from "../support/helpers/workspace-tabs"; @@ -108,7 +110,207 @@ async function withPaseoOwnedWorktree( } } +async function readCachedRowIds( + page: Page, + rowKind: "workspace" | "checkpoint", +): Promise { + return page.evaluate( + async ({ serverId, kind }) => { + const database = await new Promise((resolve, reject) => { + const request = indexedDB.open("paseo-replica-row-store"); + request.addEventListener("success", () => resolve(request.result)); + request.addEventListener("error", () => reject(request.error)); + }); + try { + return await new Promise((resolve, reject) => { + const transaction = database.transaction("rows", "readonly"); + const request = transaction.objectStore("rows").getAll(); + request.onsuccess = () => + resolve( + request.result + .filter( + (row: { serverId: string; kind: string }) => + row.serverId === serverId && row.kind === kind, + ) + .map((row: { id: string }) => row.id) + .sort(), + ); + request.addEventListener("error", () => reject(request.error)); + }); + } finally { + database.close(); + } + }, + { serverId: getServerId(), kind: rowKind }, + ); +} + +async function simulateDamagedLegacyDirectoryCache( + page: Page, + workspaceIdToRemove: string, +): Promise { + // Stop the app before altering storage so no pending app write can repair the fixture. + const fixtureUrl = new URL("/__directory_cache_fixture", page.url()).href; + await page.route(fixtureUrl, (route) => + route.fulfill({ contentType: "text/html", body: "Cache fixture" }), + ); + await page.goto(fixtureUrl); + await page.evaluate( + async ({ serverId, missingWorkspaceId }) => { + const database = await new Promise((resolve, reject) => { + const request = indexedDB.open("paseo-replica-row-store"); + request.addEventListener("success", () => resolve(request.result)); + request.addEventListener("error", () => reject(request.error)); + }); + try { + await new Promise((resolve, reject) => { + const transaction = database.transaction("rows", "readwrite"); + const store = transaction.objectStore("rows"); + const request = store.getAll(); + request.onsuccess = () => { + const rows = request.result as Array<{ + serverId: string; + kind: string; + id: string; + payload: string; + }>; + const checkpoint = rows.find( + (row) => row.serverId === serverId && row.kind === "checkpoint", + ); + if (!checkpoint) { + transaction.abort(); + return; + } + const stored = JSON.parse(checkpoint.payload); + // Accept either writer so this fixture also reproduces on the unfixed client. + const cursors = stored.version === 1 ? stored.cursors : stored; + if (!cursors.workspaces?.generation) { + transaction.abort(); + return; + } + store.put({ ...checkpoint, payload: JSON.stringify(cursors) }); + for (const row of rows) { + if ( + row.serverId === serverId && + row.kind === "workspace" && + row.id === missingWorkspaceId + ) + store.delete([serverId, row.kind, row.id]); + } + }; + transaction.addEventListener("complete", () => resolve()); + transaction.addEventListener("abort", () => + reject(transaction.error ?? new Error("Missing current directory checkpoint")), + ); + transaction.addEventListener("error", () => reject(transaction.error)); + }); + } finally { + database.close(); + } + }, + { serverId: getServerId(), missingWorkspaceId: workspaceIdToRemove }, + ); + expect(await readCachedRowIds(page, "workspace")).not.toContain(workspaceIdToRemove); +} + +async function holdHostResponsesUntilCacheIsVisible(page: Page): Promise<() => void> { + let release!: () => void; + const ready = new Promise((resolve) => { + release = resolve; + }); + await page.routeWebSocket(new RegExp(`:${getE2EDaemonPort()}(?:/|$)`), (socket) => { + const server = socket.connectToServer(); + server.onMessage(async (message) => { + await ready; + socket.send(message); + }); + }); + return release; +} + +async function createPinnedSiblingWorkspace( + project: Awaited>, +): Promise { + const sibling = await project.client.createWorkspace({ + source: { kind: "directory", path: project.repoPath, projectId: project.projectId }, + title: "Pinned sibling", + }); + if (!sibling.workspace) throw new Error(sibling.error ?? "Sibling workspace was not created"); + await project.client.setWorkspacePinned(sibling.workspace.id, true); + return sibling.workspace.id; +} + +async function expectSidebarWorkspaces(page: Page, workspaceIds: string[]): Promise { + for (const workspaceId of workspaceIds) { + await waitForSidebarWorkspace(page, workspaceId); + } +} + +async function cacheWorkspaces(page: Page, workspaceIds: string[]): Promise { + await gotoAppShell(page); + await expectSidebarWorkspaces(page, workspaceIds); + await expect.poll(() => readCachedRowIds(page, "workspace")).toEqual(workspaceIds); +} + +async function reopenDamagedCacheBeforeHostResponds( + page: Page, + missingWorkspaceId: string, +): Promise<() => void> { + await simulateDamagedLegacyDirectoryCache(page, missingWorkspaceId); + const releaseHost = await holdHostResponsesUntilCacheIsVisible(page); + await gotoAppShell(page); + return releaseHost; +} + +async function expectDamagedCacheHydrated( + page: Page, + retainedWorkspaceId: string, + missingWorkspaceId: string, +): Promise { + await waitForSidebarWorkspace(page, retainedWorkspaceId); + await expect(page.getByTestId(getWorkspaceRowTestId(missingWorkspaceId))).toHaveCount(0); + await expect.poll(() => readCachedRowIds(page, "checkpoint")).toEqual([]); +} + +async function expectRecoveredWorkspacesAfterReload( + page: Page, + workspaceIds: string[], + testInfo: TestInfo, +): Promise { + await expectSidebarWorkspaces(page, workspaceIds); + await expect.poll(() => readCachedRowIds(page, "workspace")).toEqual(workspaceIds); + await page.reload(); + await expectSidebarWorkspaces(page, workspaceIds); + await page.screenshot({ path: testInfo.outputPath("recovered-workspaces.png") }); + await testInfo.attach("recovered-workspaces", { + path: testInfo.outputPath("recovered-workspaces.png"), + contentType: "image/png", + }); +} + test.describe("Sidebar workspace list", () => { + test("recovers missing workspaces from a damaged legacy cache after reload", async ({ + page, + }, testInfo) => { + const project = await seedWorkspace({ repoPrefix: "directory-cache-recovery-" }); + try { + const siblingId = await createPinnedSiblingWorkspace(project); + const ids = [project.workspaceId, siblingId].sort(); + await test.step("cache both workspaces", () => cacheWorkspaces(page, ids)); + + const releaseHost = await test.step("reopen damaged cache before host responds", () => + reopenDamagedCacheBeforeHostResponds(page, project.workspaceId)); + await test.step("keep cached workspace and discard stale checkpoint", () => + expectDamagedCacheHydrated(page, siblingId, project.workspaceId)); + + releaseHost(); + await test.step("recover missing workspace and retain both after reload", () => + expectRecoveredWorkspacesAfterReload(page, ids, testInfo)); + } finally { + await project.cleanup(); + } + }); + test("project with GitHub remote shows its selected folder name in sidebar", async ({ page }) => { const workspace = await seedWorkspace({ repoPrefix: "sidebar-remote-", diff --git a/packages/app/src/runtime/directory-sync/index.test.ts b/packages/app/src/runtime/directory-sync/index.test.ts index ad5212ff4e..22c6c4f040 100644 --- a/packages/app/src/runtime/directory-sync/index.test.ts +++ b/packages/app/src/runtime/directory-sync/index.test.ts @@ -44,6 +44,7 @@ class FakeDirectoryClient { projectResult: ProjectListResult | null = null; private pendingAgentFetch: Promise | null = null; private pendingWorkspaceFetch: Promise | null = null; + private pendingProjectFetch: Promise | null = null; private readonly handlers = new Map< SessionOutboundMessage["type"], Set<(message: SessionOutboundMessage) => void> @@ -82,6 +83,14 @@ class FakeDirectoryClient { return complete; } + holdProjectFetch(): (result: ProjectListResult) => void { + let complete!: (result: ProjectListResult) => void; + this.pendingProjectFetch = new Promise((resolve) => { + complete = resolve; + }); + return complete; + } + async fetchAgents(options?: unknown): Promise>> { this.fetchAgentsCalls += 1; this.lastAgentOptions = options; @@ -116,6 +125,11 @@ class FakeDirectoryClient { async listProjects(options?: unknown): Promise { this.listProjectsCalls += 1; this.lastProjectOptions = options; + if (this.pendingProjectFetch) { + const pending = this.pendingProjectFetch; + this.pendingProjectFetch = null; + return pending; + } if (this.projectResult) return this.projectResult; return { requestId: "projects", @@ -227,6 +241,25 @@ function createAgent(serverId: string, id: string) { }; } +function createWorkspaceEntry(id: string): WorkspaceFetchResult["entries"][number] { + return { + id, + projectId: "snapshot-project", + projectDisplayName: "Snapshot project", + projectRootPath: "/repo", + workspaceDirectory: `/repo/${id}`, + projectKind: "git", + workspaceKind: "local_checkout", + name: id, + status: "done", + statusEnteredAt: null, + activityAt: null, + archivingAt: null, + diffStat: null, + scripts: [], + }; +} + afterEach(() => { for (const serverId of serverIds) useSessionStore.getState().clearSession(serverId); serverIds.clear(); @@ -1115,6 +1148,41 @@ describe("DirectorySync session readiness", () => { directory.dispose(); }); + it("keeps the accepted baseline when an old connection finishes its workspace request", async () => { + const serverId = "superseded-workspace-baseline"; + const { client, directory } = createDirectory(serverId); + const store = useSessionStore.getState(); + store.initializeSession(serverId, client as unknown as DaemonClient, 1); + store.updateSessionServerInfo(serverId, { + serverId, + hostname: null, + version: "test", + features: { workspaceMultiplicity: true, directorySync: true }, + }); + directory.acceptWorkspaces([normalizeWorkspaceDescriptor(createWorkspaceEntry("retained"))]); + const release = client.holdWorkspaceFetch(); + const refresh = directory.refreshWorkspaces(); + const rejected = expect(refresh).rejects.toBeInstanceOf(DirectoryRefreshSupersededError); + await expect.poll(() => client.fetchWorkspacesCalls).toBe(1); + directory.connectionChanged({ + client: client as unknown as DaemonClient, + status: "online", + source: { clientGeneration: 1, connectionEpoch: 2 }, + }); + release({ + requestId: "old-empty-snapshot", + entries: [], + emptyProjects: [], + pageInfo: { hasMore: false, nextCursor: null, prevCursor: null }, + sync: { generation: "g", headSeq: 7, mode: "snapshot", removals: [] }, + }); + await rejected; + expect([...useSessionStore.getState().sessions[serverId].workspaces.keys()]).toEqual([ + "retained", + ]); + directory.dispose(); + }); + it("rejects a session wait on disconnect so the reconnect can refresh", async () => { const serverId = "session-wait-reconnect"; const { client, directory } = createDirectory(serverId); @@ -1182,9 +1250,19 @@ describe("DirectorySync session readiness", () => { }, }, }); + client.emit({ + type: "workspace_update", + payload: { + kind: "upsert", + workspace: { ...createWorkspaceEntry("updated-workspace"), name: "Live name" }, + }, + }); completeFetch({ requestId: "workspaces", - entries: [], + entries: [ + createWorkspaceEntry("removed-workspace"), + createWorkspaceEntry("updated-workspace"), + ], emptyProjects: [ { projectId: "snapshot-project", @@ -1207,6 +1285,11 @@ describe("DirectorySync session readiness", () => { expect(projects?.get("workspace-project")).toMatchObject({ projectDisplayName: "Project from workspace update", }); + expect( + [...useSessionStore.getState().sessions[serverId].workspaces.values()].map( + ({ id, name }) => ({ id, name }), + ), + ).toEqual([{ id: "updated-workspace", name: "Live name" }]); directory.dispose(); }); @@ -1278,6 +1361,8 @@ function createSqliteCache() { return { cache, database, + storage, + reopen: () => new ReplicaCache(storage, { clearLegacyCache: async () => {} }), holdRead: (operation: () => Promise) => { beforeRead = operation; }, @@ -1473,3 +1558,198 @@ describe("DirectorySync workspace labels", () => { directory.dispose(); }); }); + +it.each(["before", "during", "before-metadata", "damaged"] as const)( + "retains unchanged workspaces across route/full demand and restart (cache: %s)", + async (timing) => { + const cacheFirst = timing === "before"; + const damaged = timing === "damaged"; + const serverId = `cache-route-overlap-${timing}`; + serverIds.add(serverId); + const { cache, database, reopen, storage } = createSqliteCache(); + cache.setHosts([serverId]); + const projectEntry: ProjectListResult["projects"][number] = { + projectId: "P", + projectDisplayName: "Project P", + projectRootPath: "/repo", + projectKind: "git", + }; + const project = normalizeProjectDescriptor(projectEntry); + const workspaceEntries: WorkspaceFetchResult["entries"] = ["A", "B"].map((id) => ({ + id, + projectId: "P", + projectDisplayName: "Project P", + projectRootPath: "/repo", + workspaceDirectory: `/repo/${id}`, + projectKind: "git", + workspaceKind: "local_checkout", + name: id, + status: "done", + statusEnteredAt: null, + activityAt: null, + archivingAt: null, + diffStat: null, + scripts: [], + })); + const workspaces = workspaceEntries.map(normalizeWorkspaceDescriptor); + const cursor = { generation: "g", afterSeq: 7 }; + cache.replaceDirectoryBaseline(serverId, { + agents: new Map(), + projects: new Map([[project.projectId, project]]), + workspaces: new Map( + (damaged ? [] : workspaces).map((workspace) => [workspace.id, workspace]), + ), + checkpoint: { projects: cursor, workspaces: cursor, agents: cursor }, + }); + await cache.flush(); + if (damaged) { + await storage.apply({ + deletes: [], + upserts: [ + { + serverId, + kind: "checkpoint", + id: "singleton", + payload: JSON.stringify({ projects: cursor, workspaces: cursor, agents: cursor }), + }, + ], + }); + } + + // The host keeps A/B unchanged. A current cursor legitimately receives no rows. + class UnchangedDirectoryClient extends FakeDirectoryClient { + readonly workspaceSyncRequests: unknown[] = []; + override async fetchWorkspaces( + options?: Parameters[0], + ): Promise { + this.fetchWorkspacesCalls += 1; + this.lastWorkspaceOptions = options; + this.workspaceSyncRequests.push(options?.sync); + const changes = options?.sync?.generation === "g" && options.sync.afterSeq === 7; + return { + requestId: "workspaces", + entries: changes ? [] : [workspaceEntries[options?.page?.cursor ? 1 : 0]], + emptyProjects: [], + pageInfo: { + hasMore: !changes && !options?.page?.cursor, + nextCursor: !changes && !options?.page?.cursor ? "page-2" : null, + prevCursor: null, + }, + sync: { + generation: "g", + headSeq: 7, + mode: changes ? "changes" : "snapshot", + removals: [], + }, + }; + } + } + const client = new UnchangedDirectoryClient(); + const releaseAgents = client.holdAgentFetch(); + const releaseProjects = client.holdProjectFetch(); + const projectResult: ProjectListResult = { + requestId: "projects", + projects: [projectEntry], + sync: { generation: "g", headSeq: 7, mode: "snapshot", removals: [] }, + }; + client.projectResult = projectResult; + const callbacks = { + onAgentStoppedRunning: () => {}, + markAgentLoading: () => {}, + markAgentReady: () => {}, + markAgentError: () => {}, + }; + const directory = new DirectorySync(serverId, callbacks, cache); + const publishServerInfo = () => { + useSessionStore.getState().updateSessionServerInfo(serverId, { + serverId, + hostname: null, + version: "test", + features: { workspaceMultiplicity: true, directorySync: true, projectList: true }, + }); + }; + const initializeSession = (ready = true) => { + useSessionStore.getState().initializeSession(serverId, client as unknown as DaemonClient, 1); + if (ready) publishServerInfo(); + }; + const connection = { + client: client as unknown as DaemonClient, + status: "online" as const, + source: { clientGeneration: 1, connectionEpoch: 1 }, + }; + initializeSession(timing !== "before-metadata"); + directory.connectionChanged(connection); + const visibleIds = () => + [...useSessionStore.getState().sessions[serverId].workspaces.keys()].sort(); + let restarted: DirectorySync | undefined; + const restartedCache = reopen(); + try { + if (cacheFirst) await directory.restoreCachedDirectory(); + directory.setAgentRouteDemand(["agent"]); + if (timing === "before-metadata") { + await directory.restoreCachedDirectory(); + publishServerInfo(); + } + await expect.poll(() => client.listProjectsCalls).toBe(1); + await expect.poll(() => client.fetchAgentsCalls).toBe(1); + directory.setDemand({}, true); + await directory.restoreCachedDirectory(); + expect(visibleIds()).toEqual(damaged ? [] : ["A", "B"]); + + releaseProjects(projectResult); + releaseAgents({ + requestId: "agents", + entries: [], + pageInfo: { hasMore: false, nextCursor: null, prevCursor: null }, + sync: { + generation: "g", + headSeq: 7, + mode: cacheFirst ? "changes" : "snapshot", + removals: [], + }, + }); + await directory.refreshDemand(); + expect(client.lastProjectOptions).toEqual({ sync: cacheFirst ? cursor : {} }); + expect(client.workspaceSyncRequests).toEqual(cacheFirst ? [cursor] : [{}, {}]); + expect.soft(visibleIds(), "after overlapping refresh").toEqual(["A", "B"]); + await cache.flush(); + expect([...(await cache.readDirectory(serverId)).workspaces.keys()].sort()).toEqual([ + "A", + "B", + ]); + + await directory.refreshAll(); + await cache.flush(); + directory.dispose(); + useSessionStore.getState().clearSession(serverId); + restartedCache.setHosts([serverId]); + const persisted = await restartedCache.readDirectory(serverId); + expect(persisted.checkpoint?.workspaces).toEqual(cursor); + expect + .soft([...persisted.workspaces.keys()].sort(), "persisted baseline") + .toEqual(["A", "B"]); + initializeSession(); + restarted = new DirectorySync(serverId, callbacks, restartedCache); + restarted.connectionChanged(connection); + await restarted.refreshAll(); + expect.soft(visibleIds(), "after restart and refresh").toEqual(["A", "B"]); + const beforeUpdate = visibleIds(); + client.emit({ + type: "workspace_update", + payload: { kind: "upsert", workspace: workspaceEntries[0], generation: "g", seq: 8 }, + }); + expect(visibleIds()).toEqual([...new Set([...beforeUpdate, "A"])].sort()); + client.emit({ + type: "workspace_update", + payload: { kind: "upsert", workspace: workspaceEntries[1], generation: "g", seq: 9 }, + }); + expect(visibleIds()).toEqual(["A", "B"]); + } finally { + directory.dispose(); + restarted?.dispose(); + await cache.flush(); + await restartedCache.flush(); + database.close(); + } + }, +); diff --git a/packages/app/src/runtime/directory-sync/index.ts b/packages/app/src/runtime/directory-sync/index.ts index da181ee473..0f10d7f893 100644 --- a/packages/app/src/runtime/directory-sync/index.ts +++ b/packages/app/src/runtime/directory-sync/index.ts @@ -61,6 +61,10 @@ interface AgentSnapshot { syncRemovals: Array<{ id: string; seq: number }>; } +interface WorkspaceRefreshSnapshot extends WorkspaceDirectorySnapshot { + readonly requestCursors: Readonly>; +} + interface AgentPageInfo { hasMore?: boolean; hasMoreAfter?: boolean; @@ -133,7 +137,7 @@ export class DirectorySync { AgentDirectoryDelta >(); private readonly workspaceTransactions = new DirectoryTransactionOwner< - WorkspaceDirectorySnapshot, + WorkspaceRefreshSnapshot, WorkspaceDirectoryDelta >(); private readonly agents: AgentDirectoryReplica; @@ -478,14 +482,20 @@ export class DirectorySync { const onlineConnection = this.getOnlineConnection(); if (!onlineConnection) return; const { client, source } = onlineConnection; - const transaction = this.workspaceTransactions.begin(source, () => ({ - workspaces: new Map(useSessionStore.getState().sessions[this.serverId]?.workspaces), - projects: new Map(useSessionStore.getState().sessions[this.serverId]?.projects), - syncCursors: {}, - syncModes: {}, - touchedWorkspaceIds: new Set(), - touchedProjectIds: new Set(), - })); + const transaction = this.workspaceTransactions.begin(source, () => { + const baseline = this.workspaces.snapshot(); + return { + workspaces: new Map(baseline.workspaces), + projects: new Map(baseline.projects), + // Cache hydration can advance the live replica while these requests await responses. + // Every page must use the cursors belonging to the maps captured here. + requestCursors: { ...this.cursors }, + syncCursors: {}, + syncModes: {}, + touchedWorkspaceIds: new Set(), + touchedProjectIds: new Set(), + }; + }); try { await this.waitForSessionMetadata(client, source); const serverInfo = useSessionStore.getState().sessions[this.serverId]?.serverInfo; @@ -517,7 +527,7 @@ export class DirectorySync { private async fetchWorkspaceSnapshot( client: DaemonClient, source: DirectorySourceToken, - transaction: DirectoryTransaction, + transaction: DirectoryTransaction, initialSubscribe: boolean, supportsDirectorySync: boolean, ): Promise { @@ -527,7 +537,9 @@ export class DirectorySync { const query: Parameters[0] = { sort: [{ key: "activity_at", direction: "desc" }], page: cursor ? { limit: PAGE_LIMIT, cursor } : { limit: PAGE_LIMIT }, - ...(supportsDirectorySync ? { sync: this.readCursors().workspaces ?? {} } : {}), + ...(supportsDirectorySync + ? { sync: transaction.snapshot.requestCursors.workspaces ?? {} } + : {}), }; let payload: FetchWorkspacesPayload; if (subscribe) { @@ -667,7 +679,7 @@ export class DirectorySync { private assertWorkspaceTransactionCurrent( client: DaemonClient, source: DirectorySourceToken, - transaction: DirectoryTransaction, + transaction: DirectoryTransaction, ): void { if (!this.workspaceTransactions.isCurrent(transaction) || !this.isCurrent(client, source)) { throw new DirectoryRefreshSupersededError("workspace fetch no longer current"); @@ -724,11 +736,13 @@ export class DirectorySync { private async fetchProjectSnapshot( client: DaemonClient, source: DirectorySourceToken, - transaction: DirectoryTransaction, + transaction: DirectoryTransaction, supportsDirectorySync: boolean, ): Promise { const payload = await client.listProjects( - supportsDirectorySync ? { sync: this.readCursors().projects ?? {} } : undefined, + supportsDirectorySync + ? { sync: transaction.snapshot.requestCursors.projects ?? {} } + : undefined, ); this.assertWorkspaceTransactionCurrent(client, source, transaction); if (payload.sync?.mode !== "changes") transaction.snapshot.projects.clear(); @@ -753,7 +767,7 @@ export class DirectorySync { private completeWorkspaceRefresh( client: DaemonClient, source: DirectorySourceToken, - transaction: DirectoryTransaction, + transaction: DirectoryTransaction, ): void { if (!this.isCurrent(client, source) || !this.hasMatchingSession(client, source)) { throw new DirectoryRefreshSupersededError("workspace completion no longer current"); diff --git a/packages/app/src/runtime/replica-cache/index.test.ts b/packages/app/src/runtime/replica-cache/index.test.ts index 6c273a325c..bb21530a2b 100644 --- a/packages/app/src/runtime/replica-cache/index.test.ts +++ b/packages/app/src/runtime/replica-cache/index.test.ts @@ -256,6 +256,41 @@ function deleteDirectory(cache: ReplicaCache, serverId: string): void { } describe("ReplicaCache", () => { + it("rejects pre-baseline-fix checkpoints without discarding cached rows or timelines", async () => { + const storage = new MemoryStorage(); + const writer = createCache(storage); + const baseline = directory(); + writer.replaceDirectoryBaseline(SERVER_ID, baseline); + writer.commitTimeline(SERVER_ID, "agent-1", timeline()); + await writer.flush(); + await storage.apply({ + deletes: [], + upserts: [ + { + serverId: SERVER_ID, + kind: "checkpoint", + id: "singleton", + payload: JSON.stringify(baseline.checkpoint), + }, + ], + }); + + const reader = createCache(storage); + const restored = await reader.readDirectory(SERVER_ID); + expect(restored.checkpoint).toBeUndefined(); + expect([...restored.workspaces.keys()]).toEqual([...baseline.workspaces.keys()]); + expect([...restored.projects.keys()]).toEqual([...baseline.projects.keys()]); + expect([...restored.agents.keys()]).toEqual([...baseline.agents.keys()]); + expect(await reader.readTimeline(SERVER_ID, "agent-1")).toEqual(timeline()); + expect(await storage.read(SERVER_ID, ["checkpoint"])).toEqual([]); + + reader.replaceDirectoryBaseline(SERVER_ID, baseline); + await reader.flush(); + expect((await createCache(storage).readDirectory(SERVER_ID)).checkpoint).toEqual( + baseline.checkpoint, + ); + }); + it("does nothing until an owner explicitly commits data", 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 bce45d3b51..ef78112a49 100644 --- a/packages/app/src/runtime/replica-cache/index.ts +++ b/packages/app/src/runtime/replica-cache/index.ts @@ -378,6 +378,21 @@ const DirectoryCheckpointSchema = z.strictObject({ agents: DirectoryCursorSchema.optional(), }); +// Old checkpoints could describe a baseline that an overlapping refresh had discarded. +// Revalidate directory contents once without evicting cached rows or timelines. +const StoredDirectoryCheckpointSchema = z.strictObject({ + version: z.literal(1), + cursors: DirectoryCheckpointSchema, +}); + +function serializeDirectoryCheckpoint(cursors: DirectoryCheckpoint) { + return { version: 1 as const, cursors }; +} + +function deserializeDirectoryCheckpoint(payload: string): DirectoryCheckpoint { + return parseStoredPayload(StoredDirectoryCheckpointSchema, payload).cursors; +} + function deserializeTimeline(stored: StoredTimeline | null): CachedTimeline | null { if (!stored) { return null; @@ -849,7 +864,7 @@ function applyDirectoryRow( if (row.id !== REPLICA_SINGLETON_ROW_ID) { throw new Error("Replica checkpoint row id mismatch"); } - result.checkpoint = parseStoredPayload(DirectoryCheckpointSchema, row.payload); + result.checkpoint = deserializeDirectoryCheckpoint(row.payload); return; default: return; @@ -1015,7 +1030,7 @@ export class ReplicaCache { let checkpoint: DirectoryCheckpoint | undefined; if (checkpointRow) { try { - checkpoint = parseStoredPayload(DirectoryCheckpointSchema, checkpointRow.payload); + checkpoint = deserializeDirectoryCheckpoint(checkpointRow.payload); checkpoint = { ...checkpoint }; delete checkpoint[invalidEntity]; } catch { @@ -1031,7 +1046,7 @@ export class ReplicaCache { serverId: row.serverId, kind: "checkpoint", id: REPLICA_SINGLETON_ROW_ID, - payload: JSON.stringify(checkpoint), + payload: JSON.stringify(serializeDirectoryCheckpoint(checkpoint)), }, ] : [], @@ -1057,7 +1072,7 @@ export class ReplicaCache { serverId, kind: "checkpoint", id: REPLICA_SINGLETON_ROW_ID, - payload: JSON.stringify(checkpoint), + payload: JSON.stringify(serializeDirectoryCheckpoint(checkpoint)), }, ] : [], @@ -1323,7 +1338,7 @@ export class ReplicaCache { if (!value) return null; break; case "checkpoint": - value = upsert.value; + value = serializeDirectoryCheckpoint(upsert.value); break; } return {