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
This commit is contained in:
Mohamed Boudra
2026-09-26 16:58:53 +02:00
committed by GitHub
parent cb9654a65c
commit d2e2154a8a
5 changed files with 567 additions and 21 deletions
@@ -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<string[]> {
return page.evaluate(
async ({ serverId, kind }) => {
const database = await new Promise<IDBDatabase>((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<string[]>((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<void> {
// 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: "<html><body>Cache fixture</body></html>" }),
);
await page.goto(fixtureUrl);
await page.evaluate(
async ({ serverId, missingWorkspaceId }) => {
const database = await new Promise<IDBDatabase>((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<void>((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<void>((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<ReturnType<typeof seedWorkspace>>,
): Promise<string> {
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<void> {
for (const workspaceId of workspaceIds) {
await waitForSidebarWorkspace(page, workspaceId);
}
}
async function cacheWorkspaces(page: Page, workspaceIds: string[]): Promise<void> {
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<void> {
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<void> {
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-",
@@ -44,6 +44,7 @@ class FakeDirectoryClient {
projectResult: ProjectListResult | null = null;
private pendingAgentFetch: Promise<AgentFetchResult> | null = null;
private pendingWorkspaceFetch: Promise<WorkspaceFetchResult> | null = null;
private pendingProjectFetch: Promise<ProjectListResult> | 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<Awaited<ReturnType<DaemonClient["fetchAgents"]>>> {
this.fetchAgentsCalls += 1;
this.lastAgentOptions = options;
@@ -116,6 +125,11 @@ class FakeDirectoryClient {
async listProjects(options?: unknown): Promise<ProjectListResult> {
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<void>) => {
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<DaemonClient["fetchWorkspaces"]>[0],
): Promise<WorkspaceFetchResult> {
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();
}
},
);
@@ -61,6 +61,10 @@ interface AgentSnapshot {
syncRemovals: Array<{ id: string; seq: number }>;
}
interface WorkspaceRefreshSnapshot extends WorkspaceDirectorySnapshot {
readonly requestCursors: Readonly<Pick<DirectoryCheckpoint, "projects" | "workspaces">>;
}
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<WorkspaceDirectorySnapshot, WorkspaceDirectoryDelta>,
transaction: DirectoryTransaction<WorkspaceRefreshSnapshot, WorkspaceDirectoryDelta>,
initialSubscribe: boolean,
supportsDirectorySync: boolean,
): Promise<void> {
@@ -527,7 +537,9 @@ export class DirectorySync {
const query: Parameters<DaemonClient["observeWorkspaces"]>[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<WorkspaceDirectorySnapshot, WorkspaceDirectoryDelta>,
transaction: DirectoryTransaction<WorkspaceRefreshSnapshot, WorkspaceDirectoryDelta>,
): 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<WorkspaceDirectorySnapshot, WorkspaceDirectoryDelta>,
transaction: DirectoryTransaction<WorkspaceRefreshSnapshot, WorkspaceDirectoryDelta>,
supportsDirectorySync: boolean,
): Promise<void> {
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<WorkspaceDirectorySnapshot, WorkspaceDirectoryDelta>,
transaction: DirectoryTransaction<WorkspaceRefreshSnapshot, WorkspaceDirectoryDelta>,
): void {
if (!this.isCurrent(client, source) || !this.hasMatchingSession(client, source)) {
throw new DirectoryRefreshSupersededError("workspace completion no longer current");
@@ -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);
@@ -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 {