root/fusion
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(FN-2353): add structured diagnostics for remote route failures

Browse Source 
- Add remote route diagnostic helpers in dashboard routes to classify timeout, transport, and unexpected errors with consistent context fields
- Apply structured runtime logging to mesh sync settings handling and both proxy paths (SSE and wildcard) for fetch and stream failure stages
- Preserve existing HTTP behavior while enriching diagnostics with node ID, upstream path, operation stage, and normalized error metadata
- Expand mesh and proxy route test suites with runtime logger harness assertions covering structured warn/error diagnostics
This commit is contained in:
Fusion
2026-04-24 01:22:03 -07:00
committed by gsxdsm
parent 31f021a03e
commit 5e9cd88708
4 changed files with 498 additions and 25 deletions
Download Patch File Download Diff File Expand all files Collapse all files

75
packages/dashboard/src/__tests__/mesh-routes.test.ts
View File

@@ -3,6 +3,7 @@ import { EventEmitter } from "node:events";
import type { Task, TaskStore } from "@fusion/core";
import { request } from "../test-request.js";
import { createServer } from "../server.js";
import type { RuntimeLogger } from "../runtime-logger.js";
// Request helper type for the test-request module
type TestRequestFn = (
@@ -118,6 +119,38 @@ function makeNodeConfig(overrides: Partial<Record<string, unknown>> = {}) {
};
}
type RuntimeLogEntry = {
level: "info" | "warn" | "error";
scope: string;
message: string;
context?: Record<string, unknown>;
};
function createRuntimeLoggerHarness(scope = "test"): { logger: RuntimeLogger; entries: RuntimeLogEntry[] } {
const entries: RuntimeLogEntry[] = [];
const makeLogger = (currentScope: string): RuntimeLogger => ({
scope: currentScope,
info(message, context) {
entries.push({ level: "info", scope: currentScope, message, context });
},
warn(message, context) {
entries.push({ level: "warn", scope: currentScope, message, context });
},
error(message, context) {
entries.push({ level: "error", scope: currentScope, message, context });
},
child(childScope) {
return makeLogger(`${currentScope}:${childScope}`);
},
});
return {
logger: makeLogger(scope),
entries,
};
}
describe("POST /api/mesh/sync", () => {
let app: ReturnType<typeof createServer>;
@@ -525,6 +558,10 @@ describe("POST /api/mesh/sync", () => {
it("should not fail sync when settings apply fails", async () => {
const remotePayload = makeSettingsPayload({ checksum: "remote-checksum" });
const localPayload = makeSettingsPayload({ checksum: "local-checksum" });
const runtimeHarness = createRuntimeLoggerHarness();
const appWithLogger = createServer(new MockStore() as unknown as TaskStore, {
runtimeLogger: runtimeHarness.logger,
});
mockGetSettings.mockResolvedValue({});
mockGetSettingsForSync.mockResolvedValue(localPayload);
@@ -537,7 +574,7 @@ describe("POST /api/mesh/sync", () => {
});
const response = await request(
app,
appWithLogger,
"POST",
"/api/mesh/sync",
JSON.stringify({
@@ -554,15 +591,34 @@ describe("POST /api/mesh/sync", () => {
expect(response.status).toBe(200);
expect(mockMergePeers).toHaveBeenCalled();
expect((response.body as any).knownPeers).toBeDefined();
expect(runtimeHarness.entries).toContainEqual(
expect.objectContaining({
level: "warn",
scope: "test:routes:remote-route:mesh-sync",
message: "Failed to apply remote settings payload",
context: expect.objectContaining({
nodeId: "node_remote",
upstreamPath: "/api/mesh/sync",
operationStage: "apply-remote-settings",
transportClassification: "unexpected",
errorClass: "Error",
errorMessage: "Checksum mismatch",
}),
}),
);
});
it("should not fail sync when getSettingsForSync throws", async () => {
const remotePayload = makeSettingsPayload({ checksum: "remote-checksum" });
const runtimeHarness = createRuntimeLoggerHarness();
const appWithLogger = createServer(new MockStore() as unknown as TaskStore, {
runtimeLogger: runtimeHarness.logger,
});
mockGetSettings.mockRejectedValue(new Error("Settings unavailable"));
const response = await request(
app,
appWithLogger,
"POST",
"/api/mesh/sync",
JSON.stringify({
@@ -580,6 +636,21 @@ describe("POST /api/mesh/sync", () => {
expect(mockMergePeers).toHaveBeenCalled();
expect((response.body as any).knownPeers).toBeDefined();
expect((response.body as any).settings).toBeUndefined();
expect(runtimeHarness.entries).toContainEqual(
expect.objectContaining({
level: "error",
scope: "test:routes:remote-route:mesh-sync",
message: "Settings sync operation failed",
context: expect.objectContaining({
nodeId: "node_remote",
upstreamPath: "/api/mesh/sync",
operationStage: "settings-sync",
transportClassification: "unexpected",
errorClass: "Error",
errorMessage: "Settings unavailable",
}),
}),
);
});
});
});

95
packages/dashboard/src/__tests__/proxy-routes.test.ts
View File

@@ -1,6 +1,7 @@
import { describe, it, expect, vi, beforeEach, afterEach } from "vitest";
import { EventEmitter } from "node:events";
import { request, get } from "../test-request.js";
import type { RuntimeLogger } from "../runtime-logger.js";
// ── Mock @fusion/core for proxy routes ──────────────────────────────
@@ -90,6 +91,38 @@ function createMockResponse(status: number, headers: Record<string, string>, bod
};
}
type RuntimeLogEntry = {
level: "info" | "warn" | "error";
scope: string;
message: string;
context?: Record<string, unknown>;
};
function createRuntimeLoggerHarness(scope = "test"): { logger: RuntimeLogger; entries: RuntimeLogEntry[] } {
const entries: RuntimeLogEntry[] = [];
const makeLogger = (currentScope: string): RuntimeLogger => ({
scope: currentScope,
info(message, context) {
entries.push({ level: "info", scope: currentScope, message, context });
},
warn(message, context) {
entries.push({ level: "warn", scope: currentScope, message, context });
},
error(message, context) {
entries.push({ level: "error", scope: currentScope, message, context });
},
child(childScope) {
return makeLogger(`${currentScope}:${childScope}`);
},
});
return {
logger: makeLogger(scope),
entries,
};
}
// ── Tests ───────────────────────────────────────────────────────────
describe("Proxy routes", () => {
@@ -186,6 +219,10 @@ describe("Proxy routes", () => {
it("returns 502 on connection error (TypeError)", async () => {
const node = createMockRemoteNode();
mockGetNode.mockResolvedValue(node);
const runtimeHarness = createRuntimeLoggerHarness();
const appWithLogger = (await import("../server.js")).createServer(store as any, {
runtimeLogger: runtimeHarness.logger,
});
const mockFetch = vi.fn().mockRejectedValue(new TypeError("fetch failed"));
@@ -193,10 +230,25 @@ describe("Proxy routes", () => {
globalThis.fetch = mockFetch;
try {
const res = await get(app, "/api/proxy/remote-node/browse-directory");
const res = await get(appWithLogger, "/api/proxy/remote-node/browse-directory");
expect(res.status).toBe(502);
expect(res.body).toEqual({ error: "Bad Gateway" });
expect(runtimeHarness.entries).toContainEqual(
expect.objectContaining({
level: "warn",
scope: "test:routes:remote-route:proxy-wildcard",
message: "Wildcard proxy transport failure",
context: expect.objectContaining({
nodeId: "remote-node",
upstreamPath: "/browse-directory",
stage: "fetch",
transportClassification: "transport",
errorClass: "TypeError",
errorMessage: "fetch failed",
}),
}),
);
} finally {
globalThis.fetch = originalFetch;
}
@@ -205,6 +257,10 @@ describe("Proxy routes", () => {
it("returns 504 on timeout/AbortError", async () => {
const node = createMockRemoteNode();
mockGetNode.mockResolvedValue(node);
const runtimeHarness = createRuntimeLoggerHarness();
const appWithLogger = (await import("../server.js")).createServer(store as any, {
runtimeLogger: runtimeHarness.logger,
});
// Create an AbortError-like DOMException
const abortError = new DOMException("Aborted", "AbortError");
@@ -214,10 +270,24 @@ describe("Proxy routes", () => {
globalThis.fetch = mockFetch;
try {
const res = await get(app, "/api/proxy/remote-node/browse-directory");
const res = await get(appWithLogger, "/api/proxy/remote-node/browse-directory");
expect(res.status).toBe(504);
expect(res.body).toEqual({ error: "Gateway Timeout" });
expect(runtimeHarness.entries).toContainEqual(
expect.objectContaining({
level: "warn",
scope: "test:routes:remote-route:proxy-wildcard",
message: "Wildcard proxy request timed out",
context: expect.objectContaining({
nodeId: "remote-node",
upstreamPath: "/browse-directory",
stage: "fetch",
transportClassification: "timeout",
errorClass: "AbortError",
}),
}),
);
} finally {
globalThis.fetch = originalFetch;
}
@@ -438,6 +508,10 @@ describe("Proxy routes", () => {
it("returns 502 on upstream error during POST", async () => {
const node = createMockRemoteNode();
mockGetNode.mockResolvedValue(node);
const runtimeHarness = createRuntimeLoggerHarness();
const appWithLogger = (await import("../server.js")).createServer(store as any, {
runtimeLogger: runtimeHarness.logger,
});
const requestBody = JSON.stringify({ key: "value" });
const rawBody = Buffer.from(requestBody);
@@ -448,7 +522,7 @@ describe("Proxy routes", () => {
try {
const res = await request(
app,
appWithLogger,
"POST",
"/api/proxy/remote-node/settings/sync",
requestBody,
@@ -458,6 +532,21 @@ describe("Proxy routes", () => {
expect(res.status).toBe(502);
expect(res.body).toEqual({ error: "Bad Gateway" });
expect(runtimeHarness.entries).toContainEqual(
expect.objectContaining({
level: "warn",
scope: "test:routes:remote-route:proxy-wildcard",
message: "Wildcard proxy transport failure",
context: expect.objectContaining({
nodeId: "remote-node",
upstreamPath: "/settings/sync",
stage: "fetch",
transportClassification: "transport",
errorClass: "TypeError",
errorMessage: "fetch failed",
}),
}),
);
} finally {
globalThis.fetch = originalFetch;
}

120
packages/dashboard/src/__tests__/routes-proxy.test.ts
View File

@@ -3,6 +3,7 @@ import { EventEmitter } from "node:events";
import type { Task } from "@fusion/core";
import { request } from "../test-request.js";
import { createServer } from "../server.js";
import type { RuntimeLogger } from "../runtime-logger.js";
const mockInit = vi.fn().mockResolvedValue(undefined);
const mockClose = vi.fn().mockResolvedValue(undefined);
@@ -71,6 +72,38 @@ function makeNode(overrides: Partial<Record<string, unknown>> = {}) {
};
}
type RuntimeLogEntry = {
level: "info" | "warn" | "error";
scope: string;
message: string;
context?: Record<string, unknown>;
};
function createRuntimeLoggerHarness(scope = "test"): { logger: RuntimeLogger; entries: RuntimeLogEntry[] } {
const entries: RuntimeLogEntry[] = [];
const makeLogger = (currentScope: string): RuntimeLogger => ({
scope: currentScope,
info(message, context) {
entries.push({ level: "info", scope: currentScope, message, context });
},
warn(message, context) {
entries.push({ level: "warn", scope: currentScope, message, context });
},
error(message, context) {
entries.push({ level: "error", scope: currentScope, message, context });
},
child(childScope) {
return makeLogger(`${currentScope}:${childScope}`);
},
});
return {
logger: makeLogger(scope),
entries,
};
}
describe("Node proxy routes", () => {
const app = createServer(new MockStore() as any);
@@ -534,6 +567,8 @@ describe("Node proxy routes", () => {
type: "remote",
url: "http://remote:4040",
});
const runtimeHarness = createRuntimeLoggerHarness();
const appWithLogger = createServer(new MockStore() as any, { runtimeLogger: runtimeHarness.logger });
vi.stubGlobal(
"fetch",
@@ -542,11 +577,26 @@ describe("Node proxy routes", () => {
mockGetNode.mockResolvedValue(remoteNode);
const res = await request(app, "GET", "/api/proxy/node_remote/events");
const res = await request(appWithLogger, "GET", "/api/proxy/node_remote/events");
expect(res.status).toBe(502);
expect(res.body).toEqual({ error: "Remote node unreachable" });
expect(mockClose).toHaveBeenCalled();
expect(runtimeHarness.entries).toContainEqual(
expect.objectContaining({
level: "warn",
scope: "test:routes:remote-route:proxy-sse",
message: "SSE proxy transport failure",
context: expect.objectContaining({
nodeId: "node_remote",
upstreamPath: "/api/events",
stage: "fetch",
transportClassification: "transport",
errorClass: "TypeError",
errorMessage: "getaddrinfo ENOTFOUND",
}),
}),
);
});
it("returns 504 when SSE fetch throws AbortError", async () => {
@@ -556,6 +606,8 @@ describe("Node proxy routes", () => {
type: "remote",
url: "http://remote:4040",
});
const runtimeHarness = createRuntimeLoggerHarness();
const appWithLogger = createServer(new MockStore() as any, { runtimeLogger: runtimeHarness.logger });
const abortError = new Error("The user abort");
abortError.name = "AbortError";
@@ -564,11 +616,75 @@ describe("Node proxy routes", () => {
mockGetNode.mockResolvedValue(remoteNode);
const res = await request(app, "GET", "/api/proxy/node_remote/events");
const res = await request(appWithLogger, "GET", "/api/proxy/node_remote/events");
expect(res.status).toBe(504);
expect(res.body).toEqual({ error: "Remote node timeout" });
expect(mockClose).toHaveBeenCalled();
expect(runtimeHarness.entries).toContainEqual(
expect.objectContaining({
level: "warn",
scope: "test:routes:remote-route:proxy-sse",
message: "SSE proxy request timed out",
context: expect.objectContaining({
nodeId: "node_remote",
upstreamPath: "/api/events",
stage: "fetch",
transportClassification: "timeout",
errorMessage: "The user abort",
}),
}),
);
});
it("emits structured diagnostics when upstream SSE stream errors", async () => {
const remoteNode = makeNode({
id: "node_remote",
name: "remote",
type: "remote",
url: "http://remote:4040",
});
const runtimeHarness = createRuntimeLoggerHarness();
const appWithLogger = createServer(new MockStore() as any, { runtimeLogger: runtimeHarness.logger });
const failingStream = new ReadableStream({
start(controller) {
controller.enqueue(new TextEncoder().encode("data: hello\\n\\n"));
controller.error(new Error("stream exploded"));
},
});
const headers = new Headers({ "content-type": "text/event-stream" });
vi.stubGlobal(
"fetch",
vi.fn().mockResolvedValue({
ok: true,
status: 200,
headers,
body: failingStream,
} as unknown as Response),
);
mockGetNode.mockResolvedValue(remoteNode);
const res = await request(appWithLogger, "GET", "/api/proxy/node_remote/events");
expect(res.status).toBe(200);
expect(runtimeHarness.entries).toContainEqual(
expect.objectContaining({
level: "error",
scope: "test:routes:remote-route:proxy-sse",
message: "SSE proxy stream error",
context: expect.objectContaining({
nodeId: "node_remote",
upstreamPath: "/api/events",
stage: "upstream-stream",
transportClassification: "unexpected",
errorClass: "Error",
errorMessage: "stream exploded",
}),
}),
);
});
});
});

233
packages/dashboard/src/routes.ts
View File

@@ -1927,6 +1927,101 @@ export function createApiRoutes(store: TaskStore, options?: ServerOptions): Rout
return getOrCreateProjectStore(projectId);
}
type RemoteRouteErrorClassification = "timeout" | "transport" | "unexpected";
interface RemoteRouteDiagnosticInput {
route: string;
message: string;
nodeId?: string;
upstreamPath?: string;
stage?: string;
operationStage?: string;
error?: unknown;
level?: "info" | "warn" | "error";
context?: Record<string, unknown>;
}
function classifyRemoteRouteError(error: unknown): {
classification: RemoteRouteErrorClassification;
errorClass: string;
errorMessage: string;
} {
const fallbackMessage = String(error);
if (error instanceof Error) {
const errorClass = error.constructor?.name || error.name || "Error";
const errorMessage = error.message || fallbackMessage;
if (error.name === "AbortError") {
return {
classification: "timeout",
errorClass,
errorMessage,
};
}
if (error instanceof TypeError) {
return {
classification: "transport",
errorClass,
errorMessage,
};
}
return {
classification: "unexpected",
errorClass,
errorMessage,
};
}
if ((error as { name?: unknown } | null)?.name === "AbortError") {
return {
classification: "timeout",
errorClass: "AbortError",
errorMessage: fallbackMessage,
};
}
return {
classification: "unexpected",
errorClass: typeof error,
errorMessage: fallbackMessage,
};
}
function emitRemoteRouteDiagnostic(input: RemoteRouteDiagnosticInput): void {
const logger = runtimeLogger.child("remote-route").child(input.route);
const level = input.level ?? "error";
const context: Record<string, unknown> = {
...(input.nodeId !== undefined ? { nodeId: input.nodeId } : {}),
...(input.upstreamPath !== undefined ? { upstreamPath: input.upstreamPath } : {}),
...(input.stage !== undefined ? { stage: input.stage } : {}),
...(input.operationStage !== undefined ? { operationStage: input.operationStage } : {}),
...(input.context ?? {}),
};
if (input.error !== undefined) {
const classified = classifyRemoteRouteError(input.error);
context.transportClassification = classified.classification;
context.errorClass = classified.errorClass;
context.errorMessage = classified.errorMessage;
}
if (level === "info") {
logger.info(input.message, context);
return;
}
if (level === "warn") {
logger.warn(input.message, context);
return;
}
logger.error(input.message, context);
}
/**
* Forward an HTTP request to a remote node.
* Validates the node exists, is remote, and has a URL, then proxies the request.
@@ -16825,11 +16920,29 @@ async function persistImportedSkills(
const applyResult = await central.applyRemoteSettings(remoteSettings);
if (applyResult.success) {
runtimeLogger.child("mesh/sync").info(
`Applied remote settings from ${senderNodeId}: global=${applyResult.globalCount}, projects=${applyResult.projectCount}, auth=${applyResult.authCount}`,
);
emitRemoteRouteDiagnostic({
route: "mesh-sync",
message: "Applied remote settings payload",
nodeId: senderNodeId,
upstreamPath: "/api/mesh/sync",
operationStage: "apply-remote-settings",
level: "info",
context: {
globalCount: applyResult.globalCount,
projectCount: applyResult.projectCount,
authCount: applyResult.authCount,
},
});
} else {
runtimeLogger.child("mesh/sync").warn(`Failed to apply remote settings: ${applyResult.error}`);
emitRemoteRouteDiagnostic({
route: "mesh-sync",
message: "Failed to apply remote settings payload",
nodeId: senderNodeId,
upstreamPath: "/api/mesh/sync",
operationStage: "apply-remote-settings",
level: "warn",
error: new Error(applyResult.error ?? "Unknown applyRemoteSettings failure"),
});
}
}
@@ -16837,8 +16950,13 @@ async function persistImportedSkills(
responseSettings = localPayload;
} catch (err) {
// Log but don't fail the sync - peers are more important
runtimeLogger.child("mesh/sync").error("Settings sync error", {
error: err instanceof Error ? err.message : String(err),
emitRemoteRouteDiagnostic({
route: "mesh-sync",
message: "Settings sync operation failed",
nodeId: senderNodeId,
upstreamPath: "/api/mesh/sync",
operationStage: "settings-sync",
error: err,
});
}
}
@@ -18313,7 +18431,8 @@ async function persistImportedSkills(
// Parse query string and build target URL
const parsedUrl = new URL(req.url, "http://localhost");
const queryString = parsedUrl.search;
const targetUrl = new URL(`/api/events${queryString}`, node.url).toString();
const upstreamPath = `/api/events${queryString}`;
const targetUrl = new URL(upstreamPath, node.url).toString();
// Build headers, injecting Authorization if apiKey is present
const headers: Record<string, string> = {};
@@ -18359,6 +18478,14 @@ async function persistImportedSkills(
req.on("close", () => {
if (!destroyed) {
destroyed = true;
emitRemoteRouteDiagnostic({
route: "proxy-sse",
message: "Closing SSE proxy stream after client disconnect",
nodeId,
upstreamPath,
stage: "client-disconnect",
level: "info",
});
controller.abort();
nodeStream.destroy();
}
@@ -18378,29 +18505,61 @@ async function persistImportedSkills(
nodeStream.on("error", (err: Error) => {
// Log but don't crash — stream may already be closing
runtimeLogger.child("proxy:sse").error(`Stream error for node ${nodeId}`, {
error: err.message,
emitRemoteRouteDiagnostic({
route: "proxy-sse",
message: "SSE proxy stream error",
nodeId,
upstreamPath,
stage: "upstream-stream",
error: err,
});
if (!res.writableEnded) {
res.end();
}
});
} catch (err: unknown) {
const parsedUrl = new URL(req.url, "http://localhost");
const queryString = parsedUrl.search;
const upstreamPath = `/api/events${queryString}`;
if (err instanceof Error && err.name === "AbortError") {
emitRemoteRouteDiagnostic({
route: "proxy-sse",
message: "SSE proxy request timed out",
nodeId,
upstreamPath,
stage: "fetch",
error: err,
level: "warn",
});
if (!res.headersSent) {
res.status(504).json({ error: "Remote node timeout" });
} else if (!res.writableEnded) {
res.end();
}
} else if (err instanceof TypeError) {
emitRemoteRouteDiagnostic({
route: "proxy-sse",
message: "SSE proxy transport failure",
nodeId,
upstreamPath,
stage: "fetch",
error: err,
level: "warn",
});
if (!res.headersSent) {
res.status(502).json({ error: "Remote node unreachable" });
} else if (!res.writableEnded) {
res.end();
}
} else {
runtimeLogger.child("proxy:sse").error(`Unexpected error for node ${nodeId}`, {
error: err instanceof Error ? err.message : String(err),
emitRemoteRouteDiagnostic({
route: "proxy-sse",
message: "SSE proxy unexpected failure",
nodeId,
upstreamPath,
stage: "fetch",
error: err,
});
if (!res.headersSent) {
if (err instanceof ApiError) {
@@ -18525,28 +18684,66 @@ async function persistImportedSkills(
});
nodeStream.on("error", (err: Error) => {
proxyLogger.error(`Stream error for node ${nodeId}`, {
error: err.message,
emitRemoteRouteDiagnostic({
route: "proxy-wildcard",
message: "Wildcard proxy stream error",
nodeId,
upstreamPath: targetPath,
stage: "upstream-stream",
error: err,
});
if (!res.writableEnded) {
res.end();
}
});
} catch (err: unknown) {
if (res.headersSent) {
return;
}
const parsedUrl = new URL(req.url ?? "/", "http://localhost");
const queryString = parsedUrl.search;
const targetPath = `/${remainingPath}${queryString}`;
// Check for AbortError by name property (works for both native Error and jsdom DOMException)
const errorObj = err as { name?: string } | null;
const isAbortError = errorObj?.name === "AbortError";
if (isAbortError) {
emitRemoteRouteDiagnostic({
route: "proxy-wildcard",
message: "Wildcard proxy request timed out",
nodeId,
upstreamPath: targetPath,
stage: "fetch",
error: err,
level: "warn",
});
if (res.headersSent) {
return;
}
res.status(504).json({ error: "Gateway Timeout" });
} else if (err instanceof TypeError) {
emitRemoteRouteDiagnostic({
route: "proxy-wildcard",
message: "Wildcard proxy transport failure",
nodeId,
upstreamPath: targetPath,
stage: "fetch",
error: err,
level: "warn",
});
if (res.headersSent) {
return;
}
res.status(502).json({ error: "Bad Gateway" });
} else {
proxyLogger.error(`Unexpected error for node ${nodeId}`, {
error: err instanceof Error ? err.message : String(err),
emitRemoteRouteDiagnostic({
route: "proxy-wildcard",
message: "Wildcard proxy unexpected failure",
nodeId,
upstreamPath: targetPath,
stage: "fetch",
error: err,
});
if (res.headersSent) {
return;
}
res.status(502).json({ error: "Bad Gateway" });
}
} finally {