fix(workbench): share sync reads and retry failed invalidations

src/features/workbench/api/project-client.test.ts

@@ -6,7 +6,10 @@ import {
 } from "./project-client";
 
 describe("workbench project client", () => {
-  afterEach(() => vi.unstubAllGlobals());
+  afterEach(() => {
+    vi.unstubAllGlobals();
+    vi.restoreAllMocks();
+  });
 
   it("selects the active project owned by the configured GeoServer workspace", async () => {
     const fetchMock = vi.fn(async (_input: RequestInfo | URL, _init?: RequestInit) => new Response(JSON.stringify({
@@ -36,23 +39,20 @@ describe("workbench project client", () => {
     await expect(findOwnedWorkbenchProject()).rejects.toThrow("授权项目");
   });
 
-  it("does not share a caller-bound abort signal with other project consumers", async () => {
+  it("shares project lookup while cancelling only the caller that aborts", async () => {
     const projectPage = JSON.stringify({
       items: [{ project_id: "supply", gs_workspace: "tjwater_next", status: "active" }],
       total: 1,
       limit: 1000,
       offset: 0
     });
-    const fetchMock = vi.fn((_input: RequestInfo | URL, init?: RequestInit) => {
-      if (!init?.signal) {
-        return Promise.resolve(new Response(projectPage, { status: 200 }));
-      }
-      return new Promise<Response>((_resolve, reject) => {
-        init.signal?.addEventListener("abort", () => {
-          reject(new DOMException("Aborted", "AbortError"));
-        }, { once: true });
-      });
-    });
+    let resolveFetch!: (response: Response) => void;
+    const fetchMock = vi.fn((_input: RequestInfo | URL, init?: RequestInit) =>
+      new Promise<Response>((resolve, reject) => {
+        resolveFetch = resolve;
+        init?.signal?.addEventListener("abort", () => reject(new DOMException("Aborted", "AbortError")), { once: true });
+      })
+    );
     vi.stubGlobal("fetch", fetchMock);
     const getAccessToken = async () => null;
     const controller = new AbortController();
@@ -63,8 +63,32 @@ describe("workbench project client", () => {
     controller.abort();
 
     await expect(cancelled).rejects.toMatchObject({ name: "AbortError" });
+    resolveFetch(new Response(projectPage, { status: 200 }));
     await expect(active).resolves.toMatchObject({ project: { project_id: "supply" } });
+    expect(fetchMock).toHaveBeenCalledTimes(1);
+  });
+
+  it("refreshes tokens, expires project metadata and evicts failed shared lookups", async () => {
+    const clock = vi.spyOn(Date, "now").mockReturnValue(0);
+    const page = { items: [{ project_id: "supply", gs_workspace: "tjwater_next", status: "active" }],
+      total: 1, limit: 1000, offset: 0 };
+    const fetchMock = vi.fn()
+      .mockResolvedValueOnce(new Response(JSON.stringify({ detail: "temporary" }), { status: 503 }))
+      .mockImplementation(async () => new Response(JSON.stringify(page), { status: 200 }));
+    vi.stubGlobal("fetch", fetchMock);
+    let token = "first";
+    const provider = vi.fn(async () => token);
+    await expect(findOwnedWorkbenchProject(provider)).rejects.toThrow("temporary");
+    await findOwnedWorkbenchProject(provider);
+    token = "renewed";
+    expect((await findOwnedWorkbenchProject(provider)).accessToken).toBe("renewed");
     expect(fetchMock).toHaveBeenCalledTimes(2);
+    clock.mockReturnValue(5 * 60_000);
+    await findOwnedWorkbenchProject(provider);
+    expect(fetchMock).toHaveBeenCalledTimes(3);
+    expect(new Headers(fetchMock.mock.calls[2][1].headers).get("Authorization")).toBe("Bearer renewed");
+    await findOwnedWorkbenchProject(async () => "another-identity");
+    expect(fetchMock).toHaveBeenCalledTimes(4);
   });
 
   it("loads the authoritative project extent and access identity independently", async () => {

src/features/workbench/api/project-client.ts

@@ -69,22 +69,39 @@ export type WorkbenchProjectPresentation = {
 };
 
 const SERVER_API_BASE_URL = env.TJWATER_SERVER_API_BASE_URL.replace(/\/$/, "");
-const projectContextPromises = new WeakMap<AccessTokenProvider, Promise<WorkbenchProject>>();
+const PROJECT_CACHE_TTL_MS = 5 * 60_000;
+const projectContextPromises = new WeakMap<AccessTokenProvider, {
+  promise: Promise<WorkbenchProject>;
+  expiresAt: number;
+}>();
+
+function awaitProject(promise: Promise<WorkbenchProject>, signal?: AbortSignal) {
+  if (!signal) return promise;
+  return new Promise<WorkbenchProject>((resolve, reject) => {
+    const abort = () => reject(signal.reason ?? new DOMException("Aborted", "AbortError"));
+    if (signal.aborted) { abort(); return; }
+    signal.addEventListener("abort", abort, { once: true });
+    promise.then(resolve, reject).finally(() => signal.removeEventListener("abort", abort));
+  });
+}
 
 export async function findOwnedWorkbenchProject(
   getAccessToken?: AccessTokenProvider,
   signal?: AbortSignal
 ): Promise<{ project: WorkbenchProject; accessToken: string | null | undefined }> {
   const accessToken = await getAccessToken?.();
-  const cacheable = Boolean(getAccessToken && !signal);
-  const existing = cacheable && getAccessToken
+  signal?.throwIfAborted();
+  const existing = getAccessToken
     ? projectContextPromises.get(getAccessToken)
     : undefined;
-  if (existing) return { project: await existing, accessToken };
+  if (existing && existing.expiresAt > Date.now()) {
+    return { project: await awaitProject(existing.promise, signal), accessToken };
+  }
   const loadProject = async () => {
     const response = await fetch(`${SERVER_API_BASE_URL}/api/v1/projects?limit=1000&offset=0`, {
       headers: createWorkbenchHeaders(accessToken),
-      signal
+      // One caller cancelling must not cancel other consumers of this lookup.
+      signal: getAccessToken ? AbortSignal.timeout(15_000) : signal
     });
     const payload: unknown = await response.json().catch(() => null);
     if (!response.ok) throw new Error(getServerError(payload, response.status, "项目查询"));
@@ -96,15 +113,13 @@ export async function findOwnedWorkbenchProject(
     return project;
   };
   const pending = loadProject();
-  if (cacheable && getAccessToken) projectContextPromises.set(getAccessToken, pending);
-  try {
-    return { project: await pending, accessToken };
-  } catch (error) {
-    if (cacheable && getAccessToken && projectContextPromises.get(getAccessToken) === pending) {
-      projectContextPromises.delete(getAccessToken);
-    }
-    throw error;
+  if (getAccessToken) {
+    projectContextPromises.set(getAccessToken, { promise: pending, expiresAt: Date.now() + PROJECT_CACHE_TTL_MS });
+    void pending.catch(() => {
+      if (projectContextPromises.get(getAccessToken)?.promise === pending) projectContextPromises.delete(getAccessToken);
+    });
   }
+  return { project: await awaitProject(pending, signal), accessToken };
 }
 
 export async function fetchWorkbenchProjectPresentation(

src/features/workbench/api/workbench-sync-client.test.ts

@@ -16,6 +16,16 @@ vi.mock("./project-client", () => ({
 afterEach(() => vi.unstubAllGlobals());
 
 describe("workbench event stream", () => {
+  it("does not report a failed HTTP connection as open", async () => {
+    vi.stubGlobal("fetch", vi.fn(async () => new Response("{}", { status: 503 })));
+    const onOpen = vi.fn();
+    await expect(consumeWorkbenchEvents({
+      after: 0, signal: new AbortController().signal, onOpen,
+      onEvents: vi.fn(), onReset: vi.fn(async () => undefined)
+    })).rejects.toThrow();
+    expect(onOpen).not.toHaveBeenCalled();
+  });
+
   it("sends the authenticated cursor and stops for a snapshot reset", async () => {
     const payload = "event: reset_required\r\ndata: {}\r\n\r\n";
     const bytes = new TextEncoder().encode(payload);
@@ -32,12 +42,14 @@ describe("workbench event stream", () => {
     }));
     const onReset = vi.fn(async () => undefined);
     const onEvents = vi.fn();
+    const onOpen = vi.fn();
 
     await expect(consumeWorkbenchEvents({
       after: 41,
       signal: new AbortController().signal,
       onReset,
-      onEvents
+      onEvents,
+      onOpen
     })).resolves.toBe("reset");
 
     expect(request?.headers.get("Authorization")).toBe("Bearer test-token");
@@ -46,6 +58,7 @@ describe("workbench event stream", () => {
     expect(new URL(request!.url).searchParams.get("after")).toBe("41");
     expect(onReset).toHaveBeenCalledOnce();
     expect(onEvents).not.toHaveBeenCalled();
+    expect(onOpen).toHaveBeenCalledOnce();
   });
 
   it("delivers every event from one stream chunk as an ordered batch", async () => {

src/features/workbench/api/workbench-sync-client.ts

@@ -82,6 +82,7 @@ export async function consumeWorkbenchEvents(input: {
   after: number;
   getAccessToken?: AccessTokenProvider;
   onEvents: (events: WorkbenchEvent[]) => void;
+  onOpen?: () => void;
   onReset: () => Promise<void>;
   signal: AbortSignal;
 }): Promise<"closed" | "reset"> {
@@ -98,30 +99,36 @@ export async function consumeWorkbenchEvents(input: {
     throw new Error(getServerError(payload, response.status, "工作台事件连接"));
   }
   const reader = response.body.getReader();
+  input.onOpen?.();
   const decoder = new TextDecoder();
   let buffer = "";
-  while (!input.signal.aborted) {
-    const { value, done } = await reader.read();
-    buffer += decoder.decode(value, { stream: !done });
-    if (buffer.length > 1_048_576) throw new Error("工作台事件帧超过大小限制");
-    const events: WorkbenchEvent[] = [];
-    let boundary = buffer.search(/\r?\n\r?\n/);
-    while (boundary >= 0) {
-      const separator = buffer.slice(boundary).match(/^\r?\n\r?\n/)?.[0] ?? "\n\n";
-      const block = buffer.slice(0, boundary);
-      buffer = buffer.slice(boundary + separator.length);
-      const lines = block.split(/\r?\n/);
-      const eventName = lines.find((line) => line.startsWith("event:"))?.slice(6).trim();
-      const data = lines.filter((line) => line.startsWith("data:")).map((line) => line.slice(5).trim()).join("\n");
-      if (eventName === "reset_required") {
-        await input.onReset();
-        return "reset";
+  try {
+    while (!input.signal.aborted) {
+      const { value, done } = await reader.read();
+      buffer += decoder.decode(value, { stream: !done });
+      if (buffer.length > 1_048_576) throw new Error("工作台事件帧超过大小限制");
+      const events: WorkbenchEvent[] = [];
+      let boundary = buffer.search(/\r?\n\r?\n/);
+      while (boundary >= 0) {
+        const separator = buffer.slice(boundary).match(/^\r?\n\r?\n/)?.[0] ?? "\n\n";
+        const block = buffer.slice(0, boundary);
+        buffer = buffer.slice(boundary + separator.length);
+        const lines = block.split(/\r?\n/);
+        const eventName = lines.find((line) => line.startsWith("event:"))?.slice(6).trim();
+        const data = lines.filter((line) => line.startsWith("data:")).map((line) => line.slice(5).trim()).join("\n");
+        if (eventName === "reset_required") {
+          await input.onReset();
+          return "reset";
+        }
+        if (eventName === "workbench_event" && data) events.push(eventSchema.parse(JSON.parse(data)));
+        boundary = buffer.search(/\r?\n\r?\n/);
       }
-      if (eventName === "workbench_event" && data) events.push(eventSchema.parse(JSON.parse(data)));
-      boundary = buffer.search(/\r?\n\r?\n/);
+      if (events.length > 0) input.onEvents(events);
+      if (done) return "closed";
     }
-    if (events.length > 0) input.onEvents(events);
-    if (done) return "closed";
+    return "closed";
+  } finally {
+    await reader.cancel().catch(() => undefined);
+    reader.releaseLock();
   }
-  return "closed";
 }