diff --git a/src/features/workbench/api/project-client.test.ts b/src/features/workbench/api/project-client.test.ts index 095018d..da0be18 100644 --- a/src/features/workbench/api/project-client.test.ts +++ b/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((_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((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 () => { diff --git a/src/features/workbench/api/project-client.ts b/src/features/workbench/api/project-client.ts index 04f84f7..f2a58de 100644 --- a/src/features/workbench/api/project-client.ts +++ b/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>(); +const PROJECT_CACHE_TTL_MS = 5 * 60_000; +const projectContextPromises = new WeakMap; + expiresAt: number; +}>(); + +function awaitProject(promise: Promise, signal?: AbortSignal) { + if (!signal) return promise; + return new Promise((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( diff --git a/src/features/workbench/api/workbench-sync-client.test.ts b/src/features/workbench/api/workbench-sync-client.test.ts index 7dcefd8..5752903 100644 --- a/src/features/workbench/api/workbench-sync-client.test.ts +++ b/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 () => { diff --git a/src/features/workbench/api/workbench-sync-client.ts b/src/features/workbench/api/workbench-sync-client.ts index a4b5ccb..1f1db24 100644 --- a/src/features/workbench/api/workbench-sync-client.ts +++ b/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; 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"; } diff --git a/src/features/workbench/hooks/use-operational-tasks.ts b/src/features/workbench/hooks/use-operational-tasks.ts index a6e2662..b9ef1e3 100644 --- a/src/features/workbench/hooks/use-operational-tasks.ts +++ b/src/features/workbench/hooks/use-operational-tasks.ts @@ -75,6 +75,8 @@ export function useOperationalTasks( const [error, setError] = useState(null); const [loading, setLoading] = useState(true); const [refreshGeneration, setRefreshGeneration] = useState(0); + const [eventRetryGeneration, setEventRetryGeneration] = useState(0); + const eventRetryMsRef = useRef(1_000); const forceRefreshRef = useRef(false); const detailResetGenerationRef = useRef(sync.detailResetGeneration); @@ -128,6 +130,7 @@ export function useOperationalTasks( const events = syncEvents; if (events.length === 0) return; const controller = new AbortController(); + let retryTimer: number | undefined; const lastSequence = events.at(-1)!.event_seq; const { taskIds, deletedTaskIds, notificationsChanged } = collectOperationalTaskEventChanges(events); if (deletedTaskIds.size > 0) { @@ -154,13 +157,20 @@ export function useOperationalTasks( setTasks(tasksRef.current); } acknowledgeEventsThrough(lastSequence); + eventRetryMsRef.current = 1_000; + setError(null); }).catch((nextError) => { if (!controller.signal.aborted) { setError(nextError instanceof Error ? nextError.message : "工况任务增量刷新失败。"); + retryTimer = window.setTimeout(() => setEventRetryGeneration((current) => current + 1), eventRetryMsRef.current); + eventRetryMsRef.current = Math.min(eventRetryMsRef.current * 2, 30_000); } }); - return () => controller.abort(); - }, [acknowledgeEventsThrough, getAccessToken, syncEvents]); + return () => { + controller.abort(); + if (retryTimer !== undefined) window.clearTimeout(retryTimer); + }; + }, [acknowledgeEventsThrough, eventRetryGeneration, getAccessToken, syncEvents]); const conditions = useMemo(() => tasks.map(toOperationalTaskCondition), [tasks]); diff --git a/src/features/workbench/hooks/use-workbench-sync.test.ts b/src/features/workbench/hooks/use-workbench-sync.test.ts index 03e8498..3bade34 100644 --- a/src/features/workbench/hooks/use-workbench-sync.test.ts +++ b/src/features/workbench/hooks/use-workbench-sync.test.ts @@ -29,6 +29,17 @@ describe("workbench snapshot ordering", () => { currentCursor: 100 })).toBe(true); }); + + it("accepts an authoritative cursor reset without accepting stale concurrent responses", () => { + expect(shouldApplyWorkbenchSnapshot({ + requestGeneration: 2, latestRequestGeneration: 2, + eventCursor: 0, currentCursor: 41, resetCursor: true + })).toBe(true); + expect(shouldApplyWorkbenchSnapshot({ + requestGeneration: 1, latestRequestGeneration: 2, + eventCursor: 0, currentCursor: 41, resetCursor: true + })).toBe(false); + }); }); describe("work-order projection readiness", () => { @@ -47,6 +58,7 @@ describe("work-order projection readiness", () => { it("waits for authoritative projection events instead of fetching a premature snapshot", () => { expect(shouldRefreshWorkbenchRuntime(event)).toBe(false); + expect(shouldRefreshWorkbenchRuntime({ ...event, aggregate_type: "operational_task" })).toBe(false); expect(shouldRefreshWorkbenchRuntime({ ...event, aggregate_type: "operational_action" })).toBe(true); expect(shouldRefreshWorkbenchRuntime({ ...event, topic: "workbench.alert.changed", aggregate_type: "workbench_alert" })).toBe(true); }); diff --git a/src/features/workbench/hooks/use-workbench-sync.ts b/src/features/workbench/hooks/use-workbench-sync.ts index a918710..627c2d3 100644 --- a/src/features/workbench/hooks/use-workbench-sync.ts +++ b/src/features/workbench/hooks/use-workbench-sync.ts @@ -17,7 +17,9 @@ const RUNTIME_TOPICS = new Set([ export function shouldRefreshWorkbenchRuntime(event: WorkbenchEvent) { // Source changes wake the backend synchronizer. Its committed task events // refresh the UI after the authoritative projection is ready. - return RUNTIME_TOPICS.has(event.topic) && event.aggregate_type !== "peripheral_work_order"; + if (event.topic === "workbench.work_order.changed" + && ["peripheral_work_order", "operational_task"].includes(event.aggregate_type)) return false; + return RUNTIME_TOPICS.has(event.topic); } export type WorkbenchSyncState = { @@ -35,9 +37,10 @@ export function shouldApplyWorkbenchSnapshot(input: { latestRequestGeneration: number; eventCursor: number; currentCursor: number; + resetCursor?: boolean; }) { return input.requestGeneration === input.latestRequestGeneration - && input.eventCursor >= input.currentCursor; + && (input.resetCursor || input.eventCursor >= input.currentCursor); } export function useWorkbenchSync(getAccessToken?: AccessTokenProvider): WorkbenchSyncState { @@ -49,7 +52,7 @@ export function useWorkbenchSync(getAccessToken?: AccessTokenProvider): Workbenc const cursorRef = useRef(0); const statusRef = useRef("connecting"); const runtimeRefreshRef = useRef(undefined); - const refreshSnapshotRef = useRef<((resetTaskDetails?: boolean) => Promise) | null>(null); + const refreshSnapshotRef = useRef<((resetTaskDetails?: boolean) => Promise) | null>(null); const acknowledgeEventsThrough = useCallback((eventSequence: number) => { setEvents((current) => current.filter((event) => event.event_seq > eventSequence)); }, []); @@ -66,8 +69,19 @@ export function useWorkbenchSync(getAccessToken?: AccessTokenProvider): Workbenc let retryMs = 1_000; let initialized = false; let snapshotRequestGeneration = 0; + let resettingCursor = false; + let runtimeRetryMs = 1_000; + let runtimeDirtyCursor = 0; + let appliedSnapshotCursor = 0; - const loadSnapshot = async (resetTaskDetails = false) => { + const loadSnapshot = async (resetTaskDetails = false, resetCursor = false) => { + if (resetCursor) { + resettingCursor = true; + runtimeDirtyCursor = 0; + if (runtimeRefreshRef.current !== undefined) window.clearTimeout(runtimeRefreshRef.current); + } + // Preserve a pending reset if another refresh supersedes this request. + const authoritativeReset = resettingCursor; const requestGeneration = ++snapshotRequestGeneration; const next = await fetchWorkbenchSyncSnapshot(getAccessToken, controller.signal); if ( @@ -76,44 +90,67 @@ export function useWorkbenchSync(getAccessToken?: AccessTokenProvider): Workbenc requestGeneration, latestRequestGeneration: snapshotRequestGeneration, eventCursor: next.eventCursor, - currentCursor: cursorRef.current + currentCursor: cursorRef.current, + resetCursor: authoritativeReset }) - ) return; + ) return false; cursorRef.current = next.eventCursor; - if (resetTaskDetails) setEvents([]); + appliedSnapshotCursor = next.eventCursor; + if (resetTaskDetails || authoritativeReset) setEvents([]); setSnapshot(next); - if (resetTaskDetails) setDetailResetGeneration((current) => current + 1); + if (resetTaskDetails || authoritativeReset) setDetailResetGeneration((current) => current + 1); + resettingCursor = false; setError(null); initialized = true; + return true; }; refreshSnapshotRef.current = loadSnapshot; + const scheduleRuntimeRefresh = (delayMs: number) => { + if (runtimeRefreshRef.current !== undefined) window.clearTimeout(runtimeRefreshRef.current); + runtimeRefreshRef.current = window.setTimeout(() => { + runtimeRefreshRef.current = undefined; + void loadSnapshot().then((applied) => { + if (!applied && appliedSnapshotCursor < runtimeDirtyCursor && !controller.signal.aborted) { + scheduleRuntimeRefresh(runtimeRetryMs); + runtimeRetryMs = Math.min(runtimeRetryMs * 2, 30_000); + } else { + runtimeRetryMs = 1_000; + } + }).catch((nextError) => { + if (controller.signal.aborted) return; + setError(nextError instanceof Error ? nextError.message : "工作台快照刷新失败"); + scheduleRuntimeRefresh(runtimeRetryMs); + runtimeRetryMs = Math.min(runtimeRetryMs * 2, 30_000); + }); + }, delayMs); + }; + const connect = async () => { try { - if (!initialized) await loadSnapshot(); + if (!initialized || resettingCursor) await loadSnapshot(); if (controller.signal.aborted) return; - statusRef.current = "live"; - setStatus("live"); - retryMs = 1_000; await consumeWorkbenchEvents({ after: cursorRef.current, getAccessToken, signal: controller.signal, - onReset: () => loadSnapshot(true), + onOpen: () => { + statusRef.current = "live"; + setStatus("live"); + setError(null); + retryMs = 1_000; + }, + onReset: async () => { await loadSnapshot(true, true); }, onEvents: (nextEvents) => { const unseen = nextEvents.filter((event) => event.event_seq > cursorRef.current); if (unseen.length === 0) return; cursorRef.current = unseen.at(-1)!.event_seq; setEvents((current) => [...current, ...unseen]); if (unseen.some(shouldRefreshWorkbenchRuntime)) { - if (runtimeRefreshRef.current !== undefined) window.clearTimeout(runtimeRefreshRef.current); - runtimeRefreshRef.current = window.setTimeout(() => { - void loadSnapshot().catch((nextError) => { - if (!controller.signal.aborted) { - setError(nextError instanceof Error ? nextError.message : "工作台快照刷新失败"); - } - }); - }, 250); + runtimeDirtyCursor = Math.max(runtimeDirtyCursor, + ...unseen.filter(shouldRefreshWorkbenchRuntime).map((event) => event.event_seq)); + runtimeRetryMs = 1_000; + scheduleRuntimeRefresh(250); } } }); diff --git a/tests/browser/peripheral-work-order-push.e2e.ts b/tests/browser/peripheral-work-order-push.e2e.ts index bae07d6..f4e7ea5 100644 --- a/tests/browser/peripheral-work-order-push.e2e.ts +++ b/tests/browser/peripheral-work-order-push.e2e.ts @@ -1,8 +1,8 @@ -import { expect, test } from "@playwright/test"; +import { expect, test, type Page } from "@playwright/test"; import { mockWorkbenchContext } from "./support/mock-workbench-context"; import { TEST_PROJECT_ID } from "./support/mock-scada-api"; -test("源工单推送等待投影完成后自动进入预约任务,无需刷新快照", async ({ page }) => { +async function expectPushedOrder(page: Page, failFirstDetail = false) { await mockWorkbenchContext(page); const base = "https://tjwater-api.invalid/api/v1"; const taskId = "72000000-0000-4000-8000-000000000041"; @@ -53,6 +53,7 @@ test("源工单推送等待投影完成后自动进入预约任务,无需刷 }); await page.route(`${base}/operational-tasks/${taskId}`, (route) => { detailRequests += 1; + if (failFirstDetail && detailRequests === 1) return route.fulfill({ headers, status: 503, json: { detail: "temporary" } }); return route.fulfill({ headers, json: task }); }); await page.route(`${base}/workbench/events**`, async (route) => { @@ -71,7 +72,12 @@ test("源工单推送等待投影完成后自动进入预约任务,无需刷 occurred_at: now, metadata: sequence === 1 ? { source: "peripheral_sample", work_order_id: orderId } : { task_id: taskId } }; - return route.fulfill({ headers, contentType: "text/event-stream", body: `id: ${sequence}\nevent: workbench_event\ndata: ${JSON.stringify(event)}\n\n` }); + const projection = { ...event, event_seq: 3, + topic: "workbench.work_order.changed", aggregate_type: "operational_task", + event_id: "75000000-0000-4000-8000-000000000003" }; + const body = `id: ${sequence}\nevent: workbench_event\ndata: ${JSON.stringify(event)}\n\n` + + (sequence === 2 ? `id: 3\nevent: workbench_event\ndata: ${JSON.stringify(projection)}\n\n` : ""); + return route.fulfill({ headers, contentType: "text/event-stream", body }); }); const waitingForProjection = page.waitForRequest((request) => request.url().includes("/workbench/events?after=1") @@ -84,8 +90,84 @@ test("源工单推送等待投影完成后自动进入预约任务,无需刷 await waitingForProjection; expect(snapshotRequests).toBe(1); await expect(panel).not.toContainText(title); + const caughtUp = page.waitForRequest((request) => request.url().includes("/workbench/events?after=3")); releaseProjection(); await expect(panel).toContainText(title); - expect(detailRequests).toBe(1); + await caughtUp; + expect(detailRequests).toBe(failFirstDetail ? 2 : 1); expect(snapshotRequests).toBe(1); +} + +test("源工单推送等待投影完成后自动进入预约任务,无需刷新快照", async ({ page }) => { + await expectPushedOrder(page); +}); + +test("事件详情短暂失败后定向重试,无需新的业务事件或全量快照", async ({ page }) => { + await expectPushedOrder(page, true); +}); + +test("事件日志清空后的合法游标重置会从零重连,不循环请求快照", async ({ page }) => { + await mockWorkbenchContext(page); + const base = "https://tjwater-api.invalid/api/v1"; + const now = new Date().toISOString(); + const runtime = { + availability: "not_configured", source: "none", scenario_availability: "not_configured", + artifact_availability: "not_configured", schedule_availability: "not_configured", + work_order_availability: "not_configured", alert_availability: "not_configured", + generated_at: now, conditions: [], alerts: [], scenarios: [], latest_analysis: null, work_orders: [] + }; + let snapshots = 0; + const cursors: string[] = []; + const headers = { "Access-Control-Allow-Origin": "*", "Access-Control-Allow-Headers": "*" }; + await page.route(`${base}/workbench/sync-snapshot`, (route) => { + snapshots += 1; + return route.fulfill({ headers, json: { + schema: "workbench-sync-snapshot@1", event_cursor: snapshots === 1 ? 41 : 0, + runtime, tasks: [], notifications: [], generated_at: now + } }); + }); + await page.route(`${base}/workbench/events**`, (route) => { + const cursor = new URL(route.request().url()).searchParams.get("after")!; + cursors.push(cursor); + return route.fulfill({ headers, contentType: "text/event-stream", + body: cursor === "41" ? "event: reset_required\ndata: {}\n\n" : ": keepalive\n\n" }); + }); + await page.goto("/", { waitUntil: "domcontentloaded" }); + await expect.poll(() => cursors.includes("0")).toBe(true); + expect(cursors.slice(0, 2)).toEqual(["41", "0"]); + expect(snapshots).toBe(2); + await expect(page.getByRole("button", { name: "工况任务", exact: true })).toBeVisible(); +}); + +test("运行态事件的快照请求短暂失败后重试,不等待其他事件", async ({ page }) => { + await mockWorkbenchContext(page); + const base = "https://tjwater-api.invalid/api/v1"; + const now = new Date().toISOString(); + const runtime = { + availability: "not_configured", source: "none", scenario_availability: "not_configured", + artifact_availability: "not_configured", schedule_availability: "not_configured", + work_order_availability: "not_configured", alert_availability: "not_configured", + generated_at: now, conditions: [], alerts: [], scenarios: [], latest_analysis: null, work_orders: [] + }; + let snapshots = 0; + const headers = { "Access-Control-Allow-Origin": "*", "Access-Control-Allow-Headers": "*" }; + await page.route(`${base}/workbench/sync-snapshot`, (route) => { + snapshots += 1; + if (snapshots === 2) return route.fulfill({ headers, status: 503, json: { detail: "temporary" } }); + return route.fulfill({ headers, json: { + schema: "workbench-sync-snapshot@1", event_cursor: snapshots === 1 ? 41 : 42, + runtime, tasks: [], notifications: [], generated_at: now + } }); + }); + await page.route(`${base}/workbench/events**`, (route) => { + const event = { schema: "workbench-event@1", event_seq: 42, + event_id: "75000000-0000-4000-8000-000000000042", project_id: TEST_PROJECT_ID, + topic: "analysis.run.changed", aggregate_type: "analysis_run", aggregate_id: "run-1", + aggregate_revision: 1, occurred_at: now, metadata: {} }; + return route.fulfill({ headers, contentType: "text/event-stream", body: + new URL(route.request().url()).searchParams.get("after") === "41" + ? `id: 42\nevent: workbench_event\ndata: ${JSON.stringify(event)}\n\n` : ": keepalive\n\n" }); + }); + await page.goto("/", { waitUntil: "domcontentloaded" }); + await expect.poll(() => snapshots).toBe(3); });