329 lines
18 KiB
TypeScript
329 lines
18 KiB
TypeScript
import { expect, test, type Page } from "@playwright/test";
|
|
import { createNdjsonLineDecoder } from "../../src/ndjsonStream";
|
|
import { prepareApp, resultSnapshot, wideProject } from "./fixtures";
|
|
|
|
test("NDJSON preserves UTF-8, CRLF, empty lines and an unterminated final result", () => {
|
|
const events = [
|
|
{ event: "progress", message: "正在仿真 🚀", progress: 5 },
|
|
{ event: "result", result: { values: [1, -2.3e-12, null], message: "第一行\n第二行" } },
|
|
];
|
|
const bytes = new TextEncoder().encode(`\n${JSON.stringify(events[0])}\r\n\n${JSON.stringify(events[1])}`);
|
|
for (const size of [1, 2, 3, 7, 64, bytes.length]) {
|
|
const decoded: unknown[] = [];
|
|
const decoder = createNdjsonLineDecoder((line) => {
|
|
if (line.trim()) decoded.push(JSON.parse(line));
|
|
});
|
|
for (let offset = 0; offset < bytes.length; offset += size) {
|
|
decoder.write(bytes.subarray(offset, offset + size));
|
|
}
|
|
decoder.finish();
|
|
decoder.finish();
|
|
expect(decoded).toEqual(events);
|
|
}
|
|
});
|
|
|
|
test("NDJSON immediately propagates errors without consuming later events", () => {
|
|
const seen: unknown[] = [];
|
|
const decoder = createNdjsonLineDecoder((line) => seen.push(JSON.parse(line)));
|
|
expect(() => decoder.write(new TextEncoder().encode('1\ninvalid\n2\n'))).toThrow();
|
|
expect(seen).toEqual([1]);
|
|
const truncated = createNdjsonLineDecoder((line) => JSON.parse(line));
|
|
truncated.write(new TextEncoder().encode('{"event":"result"'));
|
|
expect(() => truncated.finish()).toThrow();
|
|
});
|
|
|
|
test("large NDJSON stress preserves every number across 1 KiB and 64 KiB chunks", () => {
|
|
const values = Array.from({ length: 1_000_000 }, (_, i) => Math.sin(i * 0.013) * 1e5);
|
|
const expected = { event: "result", result: { message: "大结果", values } };
|
|
const bytes = new TextEncoder().encode(JSON.stringify(expected) + "\n");
|
|
for (const size of [1024, 65536]) {
|
|
let decoded: typeof expected | undefined;
|
|
let calls = 0;
|
|
const start = performance.now();
|
|
const decoder = createNdjsonLineDecoder((line) => {
|
|
calls += 1;
|
|
decoded = JSON.parse(line);
|
|
});
|
|
for (let offset = 0; offset < bytes.length; offset += size) {
|
|
decoder.write(bytes.subarray(offset, offset + size));
|
|
}
|
|
decoder.finish();
|
|
const elapsedMs = performance.now() - start;
|
|
expect(calls).toBe(1);
|
|
expect(decoded?.result.message).toBe("大结果");
|
|
expect(decoded?.result.values.length).toBe(values.length);
|
|
// Avoid a million matcher calls; compare every sample, not just the endpoints.
|
|
expect(decoded!.result.values.every((value, index) => value === values[index])).toBe(true);
|
|
console.log(JSON.stringify({ probe: "ndjson-million-values", bytes: bytes.length, chunkBytes: size, elapsedMs }));
|
|
}
|
|
});
|
|
|
|
test("100,000 progress messages coalesce while terminal states, logs and cleanup remain immediate", async ({ page }) => {
|
|
await page.route("**/performance-probe", (route) => route.fulfill({ contentType: "text/html", body: "<title>Probe</title>" }));
|
|
await page.goto("/performance-probe");
|
|
const clockStart = new Date("2026-09-10T00:00:00Z");
|
|
await page.clock.install({ time: clockStart });
|
|
await page.clock.pauseAt(new Date(clockStart.getTime() + 1000));
|
|
await page.evaluate(async () => {
|
|
// @ts-expect-error Vite serves this browser-side TypeScript module.
|
|
const { createSimulationConsoleStore } = await import("/src/simulationConsoleStore.ts");
|
|
const initial = { state: "running", percent: 0, message: "积分中", startTime: 0, simulatedTime: 0, totalTime: 10 };
|
|
const store = createSimulationConsoleStore(initial);
|
|
const probe = { store, initial, notifications: 0 };
|
|
store.subscribe(() => { probe.notifications += 1; });
|
|
(window as any).__storeProbe = probe;
|
|
for (let i = 1; i <= 100_000; i++) store.queueProgress({ ...initial, percent: i / 1000, simulatedTime: i / 10000 });
|
|
});
|
|
expect(await page.evaluate(() => (window as any).__storeProbe.notifications)).toBe(0);
|
|
await page.clock.runFor(249);
|
|
expect(await page.evaluate(() => (window as any).__storeProbe.notifications)).toBe(0);
|
|
await page.clock.runFor(1);
|
|
expect(await page.evaluate(() => {
|
|
const p = (window as any).__storeProbe;
|
|
return [p.notifications, p.store.getSnapshot().progress.percent];
|
|
})).toEqual([1, 100]);
|
|
|
|
for (let i = 1; i <= 4; i++) {
|
|
await page.evaluate((value) => {
|
|
const p = (window as any).__storeProbe;
|
|
for (let j = 0; j < 1000; j++) p.store.queueProgress({ ...p.initial, percent: value });
|
|
}, i);
|
|
await page.clock.runFor(250);
|
|
}
|
|
expect(await page.evaluate(() => (window as any).__storeProbe.notifications)).toBe(5);
|
|
await page.evaluate(() => {
|
|
const p = (window as any).__storeProbe;
|
|
p.store.queueProgress({ ...p.initial, percent: 80 });
|
|
p.store.setProgress((current: any) => ({ ...current, message: "正在停止仿真" }));
|
|
});
|
|
await page.clock.runFor(1000);
|
|
expect(await page.evaluate(() => {
|
|
const p = (window as any).__storeProbe;
|
|
return [p.notifications, p.store.getSnapshot().progress.message, p.store.getSnapshot().progress.percent];
|
|
})).toEqual([6, "正在停止仿真", 80]);
|
|
for (const state of ["success", "warning", "error"]) {
|
|
await page.evaluate((state) => {
|
|
const p = (window as any).__storeProbe;
|
|
p.store.queueProgress({ ...p.initial, percent: 90 });
|
|
p.store.setProgress({ ...p.initial, state, message: state });
|
|
}, state);
|
|
await page.clock.runFor(500);
|
|
expect(await page.evaluate(() => (window as any).__storeProbe.store.getSnapshot().progress.state)).toBe(state);
|
|
}
|
|
const counts = await page.evaluate(() => {
|
|
const p = (window as any).__storeProbe;
|
|
for (let i = 0; i < 1000; i++) p.store.appendEntry({ id: i, timestamp: "", level: "info", message: String(i), format: "message" });
|
|
const ids = p.store.getSnapshot().entries.map((entry: any) => entry.id);
|
|
p.store.clearEntries();
|
|
p.store.queueProgress({ ...p.initial, percent: 1 });
|
|
p.store.cancelPendingProgress();
|
|
return { length: ids.length, first: ids[0], last: ids.at(-1), notifications: p.notifications };
|
|
});
|
|
expect(counts).toMatchObject({ length: 400, first: 600, last: 999 });
|
|
await page.clock.runFor(1000);
|
|
expect(await page.evaluate(() => (window as any).__storeProbe.notifications)).toBe(counts.notifications);
|
|
});
|
|
|
|
async function prepareStreamApp(page: Page, nodeCount = 4) {
|
|
await prepareApp(page);
|
|
// Instrument only the test-served module; production code has no render counters.
|
|
await page.route("**/src/App.tsx", async (route) => {
|
|
const response = await route.fetch();
|
|
const source = await response.text();
|
|
expect(source).toContain("function FlowWorkbench() {");
|
|
await route.fulfill({ response, body: source.replace("function FlowWorkbench() {", "function FlowWorkbench() { window.__workbenchRenders = (window.__workbenchRenders || 0) + 1;") });
|
|
});
|
|
await page.addInitScript(() => {
|
|
const nativeFetch = window.fetch.bind(window);
|
|
window.fetch = async (...args) => {
|
|
if (!String(args[0]).includes("/api/system-xml/simulate-stream")) return nativeFetch(...args);
|
|
const encoder = new TextEncoder();
|
|
return new Response(new ReadableStream({ start(controller) {
|
|
(window as any).__sendSimulation = (events: unknown[], close = false) => {
|
|
controller.enqueue(encoder.encode(events.map((event) => JSON.stringify(event)).join("\n") + "\n"));
|
|
if (close) controller.close();
|
|
};
|
|
(window as any).__simulationController = controller;
|
|
} }), { headers: { "Content-Type": "application/x-ndjson" } });
|
|
};
|
|
});
|
|
await page.goto("/");
|
|
const nodes = Array.from({ length: nodeCount }, (_, i) => ({
|
|
...wideProject.nodes[0], id: `generic_sensor_${i + 1}`,
|
|
position: { x: (i % 20) * 160, y: Math.floor(i / 20) * 150 },
|
|
data: { ...wideProject.nodes[0].data, label: `sensor_${i + 1}` },
|
|
}));
|
|
const project = { ...wideProject, nodes, edges: nodes.map((node, i) => ({
|
|
id: `edge_${i}`, source: node.id, target: nodes[(i + 1) % nodes.length].id,
|
|
sourceHandle: "port_b", targetHandle: "port_a", data: { isContactEdge: false },
|
|
})) };
|
|
await page.locator('input[type="file"]').setInputFiles({ name: "stress.json", mimeType: "application/json", buffer: Buffer.from(JSON.stringify(project)) });
|
|
await expect(page.locator(".react-flow__node")).toHaveCount(nodeCount);
|
|
// Let import/layout and the 600 ms autosave settle before counting simulation-only renders.
|
|
await expect(page.getByText(/已自动保存/).first()).toBeVisible();
|
|
await page.getByRole("button", { name: "运行仿真", exact: true }).click();
|
|
await expect.poll(() => page.evaluate(() => typeof (window as any).__sendSimulation)).toBe("function");
|
|
}
|
|
|
|
test("300-node workbench does not rerender for a 10,000-event progress storm", async ({ page }) => {
|
|
test.setTimeout(60_000);
|
|
const errors: string[] = [];
|
|
page.on("pageerror", (error) => errors.push(error.message));
|
|
await prepareStreamApp(page, 300);
|
|
await page.evaluate(() => (window as any).__sendSimulation([{ event: "progress", phase: "integrating", progress: 1, simulatedTime: 0.1, totalTime: 10, message: "正在积分" }]));
|
|
const consolePanel = page.getByRole("complementary", { name: "仿真控制台", exact: true });
|
|
await expect(consolePanel).toContainText("正在积分");
|
|
const before = await page.evaluate(() => (window as any).__workbenchRenders);
|
|
const started = Date.now();
|
|
for (let burst = 0; burst < 20; burst++) {
|
|
await page.evaluate((burst) => {
|
|
const events = Array.from({ length: 500 }, (_, i) => ({ event: "progress", phase: "integrating", progress: 1 + (burst * 500 + i) / 125, simulatedTime: 0.1 + (burst * 500 + i) / 1250, totalTime: 10, message: "正在积分", activitySequence: burst * 500 + i + 1 }));
|
|
(window as any).__sendSimulation(events);
|
|
return new Promise((resolve) => setTimeout(resolve, 25));
|
|
}, burst);
|
|
}
|
|
await expect(consolePanel).toContainText("81%");
|
|
const after = await page.evaluate(() => (window as any).__workbenchRenders);
|
|
expect(after).toBe(before);
|
|
await expect(consolePanel).toHaveClass(/minimized/);
|
|
await page.evaluate((result) => (window as any).__sendSimulation([{ event: "result", result }], true), resultSnapshot.result);
|
|
await expect(page.getByRole("button", { name: "运行仿真", exact: true })).toBeEnabled();
|
|
await expect(consolePanel).toContainText("仿真完成");
|
|
await expect(consolePanel).toHaveClass(/minimized/);
|
|
expect(errors).toEqual([]);
|
|
console.log(JSON.stringify({ probe: "progress-storm", nodes: 300, events: 10000, workbenchRenders: after - before, elapsedMs: Date.now() - started }));
|
|
});
|
|
|
|
test("stop and stream errors cannot be overwritten by delayed running progress", async ({ page }) => {
|
|
await prepareStreamApp(page);
|
|
let cancellationRequests = 0;
|
|
await page.route("**/api/system-xml/simulations/*/cancel", (route) => {
|
|
cancellationRequests++;
|
|
return route.fulfill({ json: { accepted: true, status: "running" } });
|
|
});
|
|
await page.evaluate(() => (window as any).__sendSimulation([
|
|
{ event: "progress", phase: "integrating", progress: 1, message: "正在积分", simulatedTime: 0.1, totalTime: 10 },
|
|
{ event: "progress", phase: "integrating", progress: 40, message: "正在积分", simulatedTime: 4, totalTime: 10 },
|
|
]));
|
|
await page.getByRole("button", { name: "停止仿真", exact: true }).click();
|
|
await expect.poll(() => cancellationRequests).toBe(1);
|
|
await page.evaluate((result) => (window as any).__sendSimulation([{ event: "result", result: { ...result, status: "stopped", success: false, partial: true, simulatedUntil: 4 } }], true), resultSnapshot.result);
|
|
const panel = page.getByRole("complementary", { name: "仿真控制台", exact: true });
|
|
await expect(panel).toContainText("已手动终止");
|
|
await page.clock.install();
|
|
await page.clock.runFor(1000);
|
|
await expect(panel).toContainText("已手动终止");
|
|
await page.getByRole("button", { name: "运行仿真", exact: true }).click();
|
|
await page.evaluate(() => (window as any).__sendSimulation([
|
|
{ event: "progress", phase: "integrating", progress: 5, message: "运行中", simulatedTime: 0.5, totalTime: 10 },
|
|
{ event: "progress", phase: "integrating", progress: 7, message: "运行中", simulatedTime: 0.7, totalTime: 10 },
|
|
{ event: "error", message: "压力测试错误", status: 422 },
|
|
], true));
|
|
await page.clock.runFor(1000);
|
|
await expect(panel).toContainText("压力测试错误");
|
|
await expect(page.getByRole("button", { name: "运行仿真", exact: true })).toBeEnabled();
|
|
});
|
|
|
|
test("browser benchmark compares old and new decoders on the same 18 MiB result", async ({ page }) => {
|
|
await page.route("**/performance-probe", (route) => route.fulfill({ contentType: "text/html", body: "<title>Probe</title>" }));
|
|
await page.goto("/performance-probe");
|
|
const measurement = await page.evaluate(async () => {
|
|
// @ts-expect-error Vite serves this browser-side TypeScript module.
|
|
const { createNdjsonLineDecoder } = await import("/src/ndjsonStream.ts");
|
|
const values = Array.from({ length: 1_000_000 }, (_, i) => Math.sin(i * 0.013) * 1e5);
|
|
const bytes = new TextEncoder().encode(JSON.stringify({ event: "result", values }) + "\n");
|
|
const chunks: Uint8Array[] = [];
|
|
for (let i = 0; i < bytes.length; i += 65536) chunks.push(bytes.subarray(i, i + 65536));
|
|
const timings: Array<{ oldMs: number; newMs: number }> = [];
|
|
let correct = true;
|
|
for (let run = 0; run < 3; run++) {
|
|
const oldDecoder = new TextDecoder();
|
|
let buffer = "";
|
|
let oldValues: number[] = [];
|
|
let started = performance.now();
|
|
for (const chunk of chunks) {
|
|
buffer += oldDecoder.decode(chunk, { stream: true });
|
|
let newlineIndex = buffer.indexOf("\n");
|
|
while (newlineIndex >= 0) {
|
|
oldValues = JSON.parse(buffer.slice(0, newlineIndex).trim()).values;
|
|
buffer = buffer.slice(newlineIndex + 1);
|
|
newlineIndex = buffer.indexOf("\n");
|
|
}
|
|
}
|
|
const oldMs = performance.now() - started;
|
|
let newValues: number[] = [];
|
|
started = performance.now();
|
|
const decoder = createNdjsonLineDecoder((line: string) => { newValues = JSON.parse(line.trim()).values; });
|
|
chunks.forEach((chunk) => decoder.write(chunk));
|
|
decoder.finish();
|
|
const newMs = performance.now() - started;
|
|
correct &&= oldValues.length === values.length && newValues.length === values.length &&
|
|
newValues.every((value, i) => value === values[i] && value === oldValues[i]);
|
|
timings.push({ oldMs, newMs });
|
|
}
|
|
return { bytes: bytes.length, chunks: chunks.length, correct, timings };
|
|
});
|
|
expect(measurement.correct).toBe(true);
|
|
// Record timings instead of a hardware-dependent millisecond assertion.
|
|
console.log(JSON.stringify({ probe: "browser-ndjson-comparison", ...measurement }));
|
|
});
|
|
|
|
test("full app publishes a million-value streamed result without opening charts", async ({ page }) => {
|
|
await prepareStreamApp(page);
|
|
const errors: string[] = [];
|
|
page.on("pageerror", (error) => errors.push(error.message));
|
|
const measurement = await page.evaluate((template) => {
|
|
const points = 10000;
|
|
const series: Record<string, number[]> = { time: Array.from({ length: points }, (_, i) => i * 10 / (points - 1)) };
|
|
for (let v = 0; v < 100; v++) series[`generic_sensor_1.value_${v}`] = Array.from({ length: points }, (_, i) => Math.sin(i * 0.013 + v) * 1e5);
|
|
const result = { ...template, diagnostics: { ...template.diagnostics, sampleCount: points }, series };
|
|
const bytes = new TextEncoder().encode(JSON.stringify({ event: "result", result }) + "\n");
|
|
(window as any).__resultStressStarted = performance.now();
|
|
for (let i = 0; i < bytes.length; i += 65536) (window as any).__simulationController.enqueue(bytes.subarray(i, i + 65536));
|
|
(window as any).__simulationController.close();
|
|
return { bytes: bytes.length, points, variables: 100 };
|
|
}, resultSnapshot.result);
|
|
await expect(page.getByRole("button", { name: "运行仿真", exact: true })).toBeEnabled();
|
|
await expect(page.getByRole("complementary", { name: "仿真控制台", exact: true })).toContainText("10000 个采样点");
|
|
expect(errors).toEqual([]);
|
|
const elapsedMs = await page.evaluate(() => performance.now() - (window as any).__resultStressStarted);
|
|
console.log(JSON.stringify({ probe: "large-result-publication", ...measurement, elapsedMs }));
|
|
});
|
|
|
|
test("activity watchdog still sees all messages and recovers a genuinely stalled run", async ({ page }) => {
|
|
await prepareStreamApp(page);
|
|
const clockStart = new Date("2026-09-10T00:00:00Z");
|
|
await page.clock.install({ time: clockStart });
|
|
await page.clock.pauseAt(new Date(clockStart.getTime() + 1000));
|
|
let cancellationRequests = 0;
|
|
await page.route("**/api/system-xml/simulations/*/cancel", (route) => {
|
|
cancellationRequests++;
|
|
return route.fulfill({ json: { accepted: true, status: "running" } });
|
|
});
|
|
await page.route(/\/api\/system-xml\/simulations\/[^/]+$/, (route) => route.fulfill({ json: {
|
|
status: "stalled", result: { ...resultSnapshot.result, status: "stalled", success: false, simulatedUntil: 5 },
|
|
} }));
|
|
const sendHeartbeat = (sequence: number) => page.evaluate((sequence) => (window as any).__sendSimulation([{
|
|
event: "progress", phase: "integrating", progress: 50, simulatedTime: 5, totalTime: 10,
|
|
message: "正在积分", heartbeat: true, activitySequence: sequence, acceptedStepSequence: 10, acceptedTime: 5,
|
|
}]), sequence);
|
|
await sendHeartbeat(1);
|
|
for (let sequence = 2; sequence <= 6; sequence++) {
|
|
await page.clock.runFor(20000);
|
|
await sendHeartbeat(sequence);
|
|
}
|
|
expect(cancellationRequests).toBe(0);
|
|
await expect(page.getByRole("button", { name: "运行仿真", exact: true })).toBeDisabled();
|
|
const panel = page.getByRole("complementary", { name: "仿真控制台", exact: true });
|
|
await expect(panel).toContainText("内部计算仍活跃");
|
|
for (let i = 0; i < 3; i++) {
|
|
await page.clock.runFor(20000);
|
|
await sendHeartbeat(6);
|
|
}
|
|
await expect.poll(() => cancellationRequests).toBe(1);
|
|
await expect(panel).toContainText("异常而终止");
|
|
await expect(page.getByRole("button", { name: "运行仿真", exact: true })).toBeEnabled();
|
|
await page.clock.runFor(1000);
|
|
await expect(panel).toContainText("异常而终止");
|
|
});
|