mirror of
https://github.com/openclaw/openclaw.git
synced 2026-11-01 04:58:23 +00:00
fix(qa): bound paired-node gateway startup readiness (#115693)
This commit is contained in:
Peter Steinberger
GitHub
parent
c29d889b9c
commit
42e9558110
1 changed files with 105 additions and 34 deletions
|
|
@ -5,6 +5,7 @@ import { once } from "node:events";
|
|||
import { createServer, type IncomingMessage, type ServerResponse } from "node:http";
|
||||
import type { AddressInfo } from "node:net";
|
||||
import path from "node:path";
|
||||
import { performance } from "node:perf_hooks";
|
||||
import { setTimeout as delay } from "node:timers/promises";
|
||||
import { GatewayClient } from "openclaw/plugin-sdk/gateway-runtime";
|
||||
import type { OpenClawPluginNodeHostCommand } from "openclaw/plugin-sdk/plugin-entry";
|
||||
|
|
@ -13,7 +14,13 @@ import { describe, expect, it, vi } from "vitest";
|
|||
import { createOllamaNodeHostCommands } from "./node-inference.js";
|
||||
|
||||
const E2E_TIMEOUT_MS = 90_000;
|
||||
const STARTUP_TIMEOUT_MS = 25_000;
|
||||
// Match the shared Gateway fixture and reserve separate bounded post-start phases.
|
||||
const STARTUP_TIMEOUT_MS = Math.floor((E2E_TIMEOUT_MS * 2) / 3);
|
||||
const CLEANUP_RESERVE_MS = 10_000;
|
||||
const INFERENCE_TIMEOUT_MS = 10_000;
|
||||
const CONNECT_ATTEMPT_TIMEOUT_MS = 5_000;
|
||||
const GATEWAY_READINESS_TIMEOUT_MS =
|
||||
E2E_TIMEOUT_MS - STARTUP_TIMEOUT_MS - INFERENCE_TIMEOUT_MS - CLEANUP_RESERVE_MS;
|
||||
const NODE_DISPLAY_NAME = "paired-ollama-e2e";
|
||||
const NODE_MODEL = "node-local:latest";
|
||||
const LOADED_NODE_MODEL = "loaded-node-local:latest";
|
||||
|
|
@ -120,6 +127,7 @@ describe("Ollama paired-node Gateway inference", () => {
|
|||
gateway.once("error", (error) => appendGatewayLog(gatewayLogs, error.message));
|
||||
|
||||
await waitForGatewayHealth(gateway, gatewayPort, gatewayLogs);
|
||||
const readinessDeadline = performance.now() + GATEWAY_READINESS_TIMEOUT_MS;
|
||||
|
||||
try {
|
||||
operator = await vi.waitFor(
|
||||
|
|
@ -127,12 +135,16 @@ describe("Ollama paired-node Gateway inference", () => {
|
|||
connectClient({
|
||||
gatewayPort,
|
||||
gatewayToken,
|
||||
readinessDeadline,
|
||||
role: "operator",
|
||||
clientName: "test",
|
||||
mode: "test",
|
||||
scopes: ["operator.admin", "operator.pairing", "operator.read", "operator.write"],
|
||||
}),
|
||||
{ timeout: 15_000, interval: 250 },
|
||||
{
|
||||
timeout: remainingPhaseTimeoutMs(readinessDeadline, "Gateway readiness"),
|
||||
interval: 250,
|
||||
},
|
||||
);
|
||||
} catch (error) {
|
||||
throw new Error(
|
||||
|
|
@ -148,6 +160,7 @@ describe("Ollama paired-node Gateway inference", () => {
|
|||
node = await connectPairedNode(operator, {
|
||||
gatewayPort,
|
||||
gatewayToken,
|
||||
readinessDeadline,
|
||||
role: "node",
|
||||
clientName: "node-host",
|
||||
mode: "node",
|
||||
|
|
@ -167,25 +180,36 @@ describe("Ollama paired-node Gateway inference", () => {
|
|||
},
|
||||
});
|
||||
|
||||
const pairedNode = await waitForPairedInferenceNode(operator, gatewayLogs);
|
||||
const pairedNode = await waitForPairedInferenceNode(
|
||||
operator,
|
||||
gatewayLogs,
|
||||
readinessDeadline,
|
||||
);
|
||||
expect(pairedNode.commands).toEqual(
|
||||
expect.arrayContaining(["ollama.models", "ollama.chat"]),
|
||||
);
|
||||
|
||||
const inferenceDeadline = performance.now() + INFERENCE_TIMEOUT_MS;
|
||||
const rejected = await invokeGatewayTool({
|
||||
port: gatewayPort,
|
||||
token: "not-the-gateway-token",
|
||||
args: { action: "discover" },
|
||||
timeoutMs: remainingPhaseTimeoutMs(inferenceDeadline, "Ollama inference"),
|
||||
});
|
||||
expect(rejected.status).toBe(401);
|
||||
|
||||
const discovered = await operator.request("node.invoke", {
|
||||
nodeId: pairedNode.nodeId,
|
||||
command: "ollama.models",
|
||||
params: {},
|
||||
timeoutMs: 15_000,
|
||||
idempotencyKey: randomUUID(),
|
||||
});
|
||||
const discoveryTimeoutMs = remainingPhaseTimeoutMs(inferenceDeadline, "Ollama discovery");
|
||||
const discovered = await operator.request(
|
||||
"node.invoke",
|
||||
{
|
||||
nodeId: pairedNode.nodeId,
|
||||
command: "ollama.models",
|
||||
params: {},
|
||||
timeoutMs: discoveryTimeoutMs,
|
||||
idempotencyKey: randomUUID(),
|
||||
},
|
||||
{ timeoutMs: discoveryTimeoutMs },
|
||||
);
|
||||
expect(discovered).toMatchObject({
|
||||
ok: true,
|
||||
nodeId: pairedNode.nodeId,
|
||||
|
|
@ -199,18 +223,23 @@ describe("Ollama paired-node Gateway inference", () => {
|
|||
},
|
||||
});
|
||||
|
||||
const chat = await operator.request("node.invoke", {
|
||||
nodeId: pairedNode.nodeId,
|
||||
command: "ollama.chat",
|
||||
params: {
|
||||
model: NODE_MODEL,
|
||||
prompt: "Reply exactly with PAIRED_NODE_OK",
|
||||
maxTokens: 16,
|
||||
timeoutMs: 10_000,
|
||||
const chatTimeoutMs = remainingPhaseTimeoutMs(inferenceDeadline, "Ollama chat");
|
||||
const chat = await operator.request(
|
||||
"node.invoke",
|
||||
{
|
||||
nodeId: pairedNode.nodeId,
|
||||
command: "ollama.chat",
|
||||
params: {
|
||||
model: NODE_MODEL,
|
||||
prompt: "Reply exactly with PAIRED_NODE_OK",
|
||||
maxTokens: 16,
|
||||
timeoutMs: chatTimeoutMs,
|
||||
},
|
||||
timeoutMs: chatTimeoutMs,
|
||||
idempotencyKey: randomUUID(),
|
||||
},
|
||||
timeoutMs: 15_000,
|
||||
idempotencyKey: randomUUID(),
|
||||
});
|
||||
{ timeoutMs: chatTimeoutMs },
|
||||
);
|
||||
expect(chat).toMatchObject({
|
||||
ok: true,
|
||||
nodeId: pairedNode.nodeId,
|
||||
|
|
@ -299,9 +328,18 @@ async function waitForGatewayHealth(
|
|||
throw new Error(`Gateway did not become healthy:\n${logs.join("")}`, { cause: lastError });
|
||||
}
|
||||
|
||||
function remainingPhaseTimeoutMs(deadline: number, phase: string): number {
|
||||
const remainingMs = Math.ceil(deadline - performance.now());
|
||||
if (remainingMs <= 0) {
|
||||
throw new Error(`${phase} exceeded its bounded E2E phase`);
|
||||
}
|
||||
return remainingMs;
|
||||
}
|
||||
|
||||
async function connectClient(params: {
|
||||
gatewayPort: number;
|
||||
gatewayToken: string;
|
||||
readinessDeadline: number;
|
||||
role: "operator" | "node";
|
||||
clientName: "test" | "node-host";
|
||||
mode: "test" | "node";
|
||||
|
|
@ -312,6 +350,11 @@ async function connectClient(params: {
|
|||
commands?: string[];
|
||||
onEvent?: (event: { event: string; payload?: unknown }) => void;
|
||||
}): Promise<GatewayClient> {
|
||||
// Leave the shared readiness deadline available for another handshake attempt.
|
||||
const timeoutMs = Math.min(
|
||||
CONNECT_ATTEMPT_TIMEOUT_MS,
|
||||
remainingPhaseTimeoutMs(params.readinessDeadline, "Gateway client readiness"),
|
||||
);
|
||||
return await new Promise<GatewayClient>((resolve, reject) => {
|
||||
let settled = false;
|
||||
const finish = (error?: Error) => {
|
||||
|
|
@ -339,7 +382,7 @@ async function connectClient(params: {
|
|||
scopes: params.scopes ?? [],
|
||||
caps: params.caps,
|
||||
commands: params.commands,
|
||||
requestTimeoutMs: 15_000,
|
||||
requestTimeoutMs: INFERENCE_TIMEOUT_MS,
|
||||
onEvent: params.onEvent,
|
||||
onHelloOk: () => finish(),
|
||||
onConnectError: (error) => finish(error),
|
||||
|
|
@ -347,29 +390,43 @@ async function connectClient(params: {
|
|||
});
|
||||
const timeout = setTimeout(
|
||||
() => finish(new Error("Gateway client connection timed out")),
|
||||
5_000,
|
||||
timeoutMs,
|
||||
);
|
||||
timeout.unref();
|
||||
client.start();
|
||||
});
|
||||
}
|
||||
|
||||
async function approvePendingNodePairings(operator: GatewayClient): Promise<void> {
|
||||
async function approvePendingNodePairings(
|
||||
operator: GatewayClient,
|
||||
readinessDeadline: number,
|
||||
): Promise<void> {
|
||||
const pairingOptions = () => ({
|
||||
timeoutMs: remainingPhaseTimeoutMs(readinessDeadline, "Gateway node pairing"),
|
||||
});
|
||||
const devices = await operator.request<{
|
||||
pending?: Array<{ requestId?: string; role?: string }>;
|
||||
}>("device.pair.list", {});
|
||||
}>("device.pair.list", {}, pairingOptions());
|
||||
for (const request of devices.pending ?? []) {
|
||||
if (request.requestId) {
|
||||
await operator.request("device.pair.approve", { requestId: request.requestId });
|
||||
await operator.request(
|
||||
"device.pair.approve",
|
||||
{ requestId: request.requestId },
|
||||
pairingOptions(),
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
const nodes = await operator.request<{
|
||||
pending?: Array<{ requestId?: string; displayName?: string }>;
|
||||
}>("node.pair.list", {});
|
||||
}>("node.pair.list", {}, pairingOptions());
|
||||
for (const request of nodes.pending ?? []) {
|
||||
if (request.requestId) {
|
||||
await operator.request("node.pair.approve", { requestId: request.requestId });
|
||||
await operator.request(
|
||||
"node.pair.approve",
|
||||
{ requestId: request.requestId },
|
||||
pairingOptions(),
|
||||
);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -387,18 +444,22 @@ async function connectPairedNode(
|
|||
}
|
||||
// The operator and node share this isolated device identity. Approving the
|
||||
// Gateway's requested role upgrade preserves the real node pairing boundary.
|
||||
await approvePendingNodePairings(operator);
|
||||
await approvePendingNodePairings(operator, params.readinessDeadline);
|
||||
return await connectClient(params);
|
||||
}
|
||||
}
|
||||
|
||||
async function waitForPairedInferenceNode(operator: GatewayClient, logs: string[]) {
|
||||
async function waitForPairedInferenceNode(
|
||||
operator: GatewayClient,
|
||||
logs: string[],
|
||||
readinessDeadline: number,
|
||||
) {
|
||||
let paired:
|
||||
| { nodeId: string; displayName?: string; connected?: boolean; commands?: string[] }
|
||||
| undefined;
|
||||
await vi.waitFor(
|
||||
async () => {
|
||||
await approvePendingNodePairings(operator);
|
||||
await approvePendingNodePairings(operator, readinessDeadline);
|
||||
const result = await operator.request<{
|
||||
nodes?: Array<{
|
||||
nodeId: string;
|
||||
|
|
@ -406,14 +467,23 @@ async function waitForPairedInferenceNode(operator: GatewayClient, logs: string[
|
|||
connected?: boolean;
|
||||
commands?: string[];
|
||||
}>;
|
||||
}>("node.list", {});
|
||||
}>(
|
||||
"node.list",
|
||||
{},
|
||||
{
|
||||
timeoutMs: remainingPhaseTimeoutMs(readinessDeadline, "Gateway node discovery"),
|
||||
},
|
||||
);
|
||||
paired = result.nodes?.find(
|
||||
(entry) => entry.displayName === NODE_DISPLAY_NAME && entry.connected,
|
||||
);
|
||||
expect(paired, logs.join("")).toBeDefined();
|
||||
expect(paired?.commands).toEqual(expect.arrayContaining(["ollama.models", "ollama.chat"]));
|
||||
},
|
||||
{ timeout: 25_000, interval: 100 },
|
||||
{
|
||||
timeout: remainingPhaseTimeoutMs(readinessDeadline, "Gateway node readiness"),
|
||||
interval: 100,
|
||||
},
|
||||
);
|
||||
if (!paired) {
|
||||
throw new Error("Ollama-capable paired node never connected");
|
||||
|
|
@ -458,6 +528,7 @@ async function invokeGatewayTool(params: {
|
|||
port: number;
|
||||
token: string;
|
||||
args: Record<string, unknown>;
|
||||
timeoutMs: number;
|
||||
}): Promise<Response> {
|
||||
return await fetch(`http://127.0.0.1:${params.port}/tools/invoke`, {
|
||||
method: "POST",
|
||||
|
|
@ -466,7 +537,7 @@ async function invokeGatewayTool(params: {
|
|||
authorization: `Bearer ${params.token}`,
|
||||
},
|
||||
body: JSON.stringify({ tool: "node_inference", args: params.args }),
|
||||
signal: AbortSignal.timeout(30_000),
|
||||
signal: AbortSignal.timeout(params.timeoutMs),
|
||||
});
|
||||
}
|
||||
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue