fix: 恢复 HarnessRL HWPOD 重启幂等
This commit is contained in:
root
@@ -700,6 +700,7 @@ lanes:
|
||||
- deploy/runtime/boot/hwlab-harnessrl-worker.sh
|
||||
env:
|
||||
TZ: Asia/Shanghai
|
||||
HWLAB_CASERUN_INTERNAL_API_URL: http://hwlab-cloud-api.hwlab-dev.svc.cluster.local:6667
|
||||
observable: true
|
||||
hwlab-cloud-web:
|
||||
runtimeKind: cloud-web
|
||||
@@ -916,6 +917,7 @@ lanes:
|
||||
timeoutMs: 2500
|
||||
env:
|
||||
HWLAB_CLOUD_DB_URL: secretRef:hwlab-cloud-api-v03-db/database-url
|
||||
HWLAB_HWPOD_OPERATION_RETENTION_SECONDS: "604800"
|
||||
HWLAB_CLOUD_DB_CONTRACT: v03-redacted-presence-only
|
||||
HWLAB_ACCESS_CONTROL_REQUIRED: "1"
|
||||
HWLAB_SESSION_COOKIE_SAMESITE: None
|
||||
@@ -1082,6 +1084,8 @@ lanes:
|
||||
HARNESSRL_ACTIVITY_RETRY_MAXIMUM_INTERVAL_MS: "30000"
|
||||
HARNESSRL_ACTIVITY_RETRY_MAXIMUM_ATTEMPTS: "5"
|
||||
HARNESSRL_WORKER_HEALTH_PORT: "6676"
|
||||
HWLAB_CASERUN_INTERNAL_API_URL: http://hwlab-cloud-api.hwlab-v03.svc.cluster.local:6667
|
||||
HWLAB_RUNTIME_INTERNAL_API_URL: http://hwlab-cloud-api.hwlab-v03.svc.cluster.local:6667
|
||||
OTEL_SERVICE_NAME: hwlab-harnessrl-worker
|
||||
services:
|
||||
- serviceId: hwlab-harnessrl-api
|
||||
|
||||
@@ -2,6 +2,7 @@ import { createCloudApiServer } from "./server.ts";
|
||||
import { createHwpodNodeWsRegistry } from "./hwpod-node-ws-registry.ts";
|
||||
import { drainWorkbenchRealtimeConnections } from "./server-workbench-http.ts";
|
||||
import { classifyRuntimeDbError } from "../db/runtime-store.ts";
|
||||
import { createConfiguredCloudRuntimeStore } from "../db/runtime-store.ts";
|
||||
|
||||
const POSTGRES_TRANSIENT_UNHANDLED_REJECTION_BOUNDARY = Symbol.for("hwlab.cloud.postgresTransientUnhandledRejectionBoundaryInstalled");
|
||||
|
||||
@@ -9,8 +10,9 @@ export async function createCloudApiBunServer(options: any = {}) {
|
||||
const env = options.env ?? process.env;
|
||||
const logger = options.logger ?? console;
|
||||
installPostgresTransientUnhandledRejectionBoundary({ env, logger });
|
||||
const hwpodNodeWsRegistry = options.hwpodNodeWsRegistry || createHwpodNodeWsRegistry({ ...options, env });
|
||||
const innerServer = createCloudApiServer({ ...options, env, hwpodNodeWsRegistry });
|
||||
const runtimeStore = options.runtimeStore || createConfiguredCloudRuntimeStore({ ...options, env });
|
||||
const hwpodNodeWsRegistry = options.hwpodNodeWsRegistry || createHwpodNodeWsRegistry({ ...options, env, runtimeStore });
|
||||
const innerServer = createCloudApiServer({ ...options, env, runtimeStore, hwpodNodeWsRegistry });
|
||||
try {
|
||||
await innerServer.hwlabStartupReady;
|
||||
} catch (error) {
|
||||
|
||||
@@ -11,6 +11,7 @@ import { test } from "bun:test";
|
||||
|
||||
import { createCloudApiBunServer } from "./bun-server.ts";
|
||||
import { createHwpodNodeWsRegistry } from "./hwpod-node-ws-registry.ts";
|
||||
import { createMemoryHwpodOperationLedger } from "./hwpod-operation-ledger.ts";
|
||||
import { createCloudApiServer } from "./server.ts";
|
||||
import { connectHwpodNodeWs, createHwpodNodeServer } from "../../tools/src/hwpod-node-lib.ts";
|
||||
|
||||
@@ -452,16 +453,57 @@ test("WebSocket registry reuses one authoritative operation for concurrent dupli
|
||||
plan.planId = "harnessrl-run:build:v1";
|
||||
const first = registry.dispatch(plan, { requestId: "req_first" }, { timeoutMs: 5000 });
|
||||
const retry = registry.dispatch(plan, { requestId: "req_retry" }, { timeoutMs: 5000 });
|
||||
await new Promise((resolve) => setTimeout(resolve, 0));
|
||||
assert.equal(dispatchCount, 1);
|
||||
registry.handleBunMessage(socket, JSON.stringify({ type: "hwpod-node-ops-result", requestId: "req_first", nodeId: "pc-host-1", result: { ok: true, status: "completed", results: [{ opId: "op_1", ok: true }] } }));
|
||||
assert.deepEqual(await retry, await first);
|
||||
assert.equal(registry.lookup(plan.planId)?.result?.status, "completed");
|
||||
assert.equal((await registry.lookup(plan.planId))?.result?.status, "completed");
|
||||
const recovered = await registry.dispatch(plan, { requestId: "req_after_completion" }, { timeoutMs: 5000 });
|
||||
assert.equal(recovered.status, "completed");
|
||||
assert.equal(dispatchCount, 1);
|
||||
registry.closeBunSocket(socket);
|
||||
});
|
||||
|
||||
test("WebSocket registry rebuild recovers one durable plan result without redispatch", async () => {
|
||||
const records = new Map();
|
||||
let dispatchCount = 0;
|
||||
const firstLedger = createMemoryHwpodOperationLedger({ records, now: () => "2026-07-17T02:40:00.000Z" });
|
||||
const firstRegistry = createHwpodNodeWsRegistry({ operationLedger: firstLedger, now: () => "2026-07-17T02:40:00.000Z" });
|
||||
const firstSocket = { send(value: string) { const message = JSON.parse(value); if (message.type !== "hwpod-node-ops") return; dispatchCount += 1; queueMicrotask(() => { firstRegistry.handleBunMessage(firstSocket, JSON.stringify({ type: "hwpod-node-ops-accepted", requestId: message.requestId, nodeId: "pc-host-1" })); firstRegistry.handleBunMessage(firstSocket, JSON.stringify({ type: "hwpod-node-ops-result", requestId: message.requestId, nodeId: "pc-host-1", result: { ok: true, status: "completed", planId: message.plan.planId, results: [] } })); }); }, close() {} };
|
||||
firstRegistry.openBunSocket(firstSocket);
|
||||
firstRegistry.handleBunMessage(firstSocket, JSON.stringify({ type: "register", nodeId: "pc-host-1", maxInFlight: 1 }));
|
||||
const plan = { ...samplePlan(), nodeId: "pc-host-1", planId: "restart-safe-plan" };
|
||||
const firstResult = await firstRegistry.dispatch(plan, { requestId: "req_first" }, { timeoutMs: 1000 });
|
||||
firstRegistry.closeBunSocket(firstSocket);
|
||||
|
||||
const rebuiltLedger = createMemoryHwpodOperationLedger({ records, now: () => "2026-07-17T02:41:00.000Z" });
|
||||
const rebuiltRegistry = createHwpodNodeWsRegistry({ operationLedger: rebuiltLedger, now: () => "2026-07-17T02:41:00.000Z" });
|
||||
const recovered = await rebuiltRegistry.dispatch(plan, { requestId: "req_rebuilt" }, { timeoutMs: 1000 });
|
||||
assert.deepEqual(recovered, firstResult);
|
||||
assert.equal((await rebuiltRegistry.lookup(plan.planId))?.result?.status, "completed");
|
||||
assert.equal(dispatchCount, 1);
|
||||
});
|
||||
|
||||
test("WebSocket registry retries the same plan identity after pre-accept send failure", async () => {
|
||||
const records = new Map();
|
||||
const plan = { ...samplePlan(), nodeId: "pc-host-1", planId: "retry-same-plan" };
|
||||
const firstRegistry = createHwpodNodeWsRegistry({ operationLedger: createMemoryHwpodOperationLedger({ records }) });
|
||||
const failedSocket = { send(value: string) { if (JSON.parse(value).type === "hwpod-node-ops") throw new Error("connection failed before acceptance"); }, close() {} };
|
||||
firstRegistry.openBunSocket(failedSocket);
|
||||
firstRegistry.handleBunMessage(failedSocket, JSON.stringify({ type: "register", nodeId: "pc-host-1", maxInFlight: 1 }));
|
||||
assert.equal((await firstRegistry.dispatch(plan, { requestId: "req_failed" }, { timeoutMs: 1000 })).status, "blocked");
|
||||
|
||||
let acceptanceCount = 0;
|
||||
const retryRegistry = createHwpodNodeWsRegistry({ operationLedger: createMemoryHwpodOperationLedger({ records }) });
|
||||
const retrySocket = { send(value: string) { const message = JSON.parse(value); if (message.type !== "hwpod-node-ops") return; acceptanceCount += 1; queueMicrotask(() => { retryRegistry.handleBunMessage(retrySocket, JSON.stringify({ type: "hwpod-node-ops-accepted", requestId: message.requestId, nodeId: "pc-host-1" })); retryRegistry.handleBunMessage(retrySocket, JSON.stringify({ type: "hwpod-node-ops-result", requestId: message.requestId, nodeId: "pc-host-1", result: { ok: true, status: "completed", planId: message.plan.planId, results: [] } })); }); }, close() {} };
|
||||
retryRegistry.openBunSocket(retrySocket);
|
||||
retryRegistry.handleBunMessage(retrySocket, JSON.stringify({ type: "register", nodeId: "pc-host-1", maxInFlight: 1 }));
|
||||
const result = await retryRegistry.dispatch(plan, { requestId: "req_retry" }, { timeoutMs: 1000 });
|
||||
assert.equal(result.status, "completed");
|
||||
assert.equal(result.planId, plan.planId);
|
||||
assert.equal(acceptanceCount, 1);
|
||||
});
|
||||
|
||||
test("WebSocket registry keeps readiness probes separate and rejects dispatch above maxInFlight", async () => {
|
||||
const sent: any[] = [];
|
||||
const socket = { send(value: string) { sent.push(JSON.parse(value)); }, close() {} };
|
||||
|
||||
@@ -4,6 +4,7 @@
|
||||
* Responsibility: 管理主动出站 HWPOD Node 连接、并发门禁，以及用户操作与 readiness 探测的独立摘要。
|
||||
*/
|
||||
import { randomUUID } from "node:crypto";
|
||||
import { createHwpodOperationLedger } from "./hwpod-operation-ledger.ts";
|
||||
|
||||
const DEFAULT_DISPATCH_TIMEOUT_MS = 30000;
|
||||
|
||||
@@ -28,6 +29,7 @@ type HwpodNodeConnection = {
|
||||
};
|
||||
|
||||
type PendingDispatch = {
|
||||
planId: string;
|
||||
nodeId: string;
|
||||
requestId: string;
|
||||
connection: HwpodNodeConnection;
|
||||
@@ -49,9 +51,16 @@ export function createHwpodNodeWsRegistry(options: any = {}) {
|
||||
const now = options.now ?? (() => new Date().toISOString());
|
||||
const clock = options.clock ?? (() => Date.now());
|
||||
const logger = options.logger ?? console;
|
||||
const env = options.env ?? process.env;
|
||||
const operationLedger = options.operationLedger ?? createHwpodOperationLedger({
|
||||
runtimeStore: options.runtimeStore,
|
||||
now,
|
||||
retentionMs: positiveInteger(env.HWLAB_HWPOD_OPERATION_RETENTION_SECONDS, 0) * 1000
|
||||
});
|
||||
const connections = new Map<string, HwpodNodeConnection>();
|
||||
const pending = new Map<string, PendingDispatch>();
|
||||
const operations = new Map<string, AuthoritativeOperation>();
|
||||
const admissions = new Map<string, { nodeId: string; promise: Promise<any> }>();
|
||||
const socketConnections = new WeakMap<object, HwpodNodeConnection>();
|
||||
|
||||
function openBunSocket(socket: any) {
|
||||
@@ -89,26 +98,51 @@ export function createHwpodNodeWsRegistry(options: any = {}) {
|
||||
}
|
||||
|
||||
function dispatch(plan: any, requestMeta: any = {}, dispatchOptions: any = {}) {
|
||||
const planId = safeText(requestMeta.operationKind) === "readiness-probe" ? "" : safeText(plan?.planId);
|
||||
const nodeId = safeNodeId(plan?.nodeId);
|
||||
const admission = planId ? admissions.get(planId) : null;
|
||||
if (admission) {
|
||||
return admission.nodeId === nodeId
|
||||
? admission.promise
|
||||
: Promise.resolve(blockedDispatch(plan, requestMeta, `hwpod operation ${planId} is already owned by node ${admission.nodeId}`, "hwpod_operation_identity_conflict"));
|
||||
}
|
||||
const promise = dispatchOperation(plan, requestMeta, dispatchOptions);
|
||||
if (planId) {
|
||||
admissions.set(planId, { nodeId, promise });
|
||||
void promise.finally(() => admissions.delete(planId));
|
||||
}
|
||||
return promise;
|
||||
}
|
||||
|
||||
async function dispatchOperation(plan: any, requestMeta: any = {}, dispatchOptions: any = {}) {
|
||||
const nodeId = safeNodeId(plan?.nodeId);
|
||||
const operationKind = safeText(requestMeta.operationKind) === "readiness-probe" ? "readiness-probe" : "user-operation";
|
||||
const planId = operationKind === "user-operation" ? safeText(plan?.planId) : "";
|
||||
const authoritative = planId ? operations.get(planId) : null;
|
||||
if (authoritative) {
|
||||
if (authoritative.nodeId !== nodeId) {
|
||||
return Promise.resolve(blockedDispatch(plan, requestMeta, `hwpod operation ${planId} is already owned by node ${authoritative.nodeId}`, "hwpod_operation_identity_conflict"));
|
||||
return blockedDispatch(plan, requestMeta, `hwpod operation ${planId} is already owned by node ${authoritative.nodeId}`, "hwpod_operation_identity_conflict");
|
||||
}
|
||||
return authoritative.result === null ? authoritative.promise : Promise.resolve(authoritative.result);
|
||||
return authoritative.result === null ? authoritative.promise : authoritative.result;
|
||||
}
|
||||
const durable = planId ? await operationLedger.get(planId) : null;
|
||||
if (durable?.nodeId && durable.nodeId !== nodeId) {
|
||||
return blockedDispatch(plan, requestMeta, `hwpod operation ${planId} is already owned by node ${durable.nodeId}`, "hwpod_operation_identity_conflict");
|
||||
}
|
||||
if (durable?.status === "completed" && durable.result) {
|
||||
return durable.result;
|
||||
}
|
||||
const connection = nodeId ? connections.get(nodeId) : null;
|
||||
if (!nodeId || !connection) {
|
||||
return Promise.resolve(blockedDispatch(plan, requestMeta, `hwpod-node ${nodeId || "<missing>"} is not connected through outbound WebSocket`));
|
||||
return blockedDispatch(plan, requestMeta, `hwpod-node ${nodeId || "<missing>"} is not connected through outbound WebSocket`);
|
||||
}
|
||||
if (inFlightCountForNode(nodeId, pending) >= connection.maxInFlight) {
|
||||
return Promise.resolve(blockedDispatch(plan, requestMeta, `hwpod-node ${nodeId} reached its maxInFlight=${connection.maxInFlight} limit`, "hwpod_node_busy"));
|
||||
return blockedDispatch(plan, requestMeta, `hwpod-node ${nodeId} reached its maxInFlight=${connection.maxInFlight} limit`, "hwpod_node_busy");
|
||||
}
|
||||
const requestId = safeText(requestMeta.requestId) || `req_hwpod_ws_${randomUUID()}`;
|
||||
const timeoutMs = positiveInteger(dispatchOptions.timeoutMs, DEFAULT_DISPATCH_TIMEOUT_MS);
|
||||
const operation = operationStarted(plan, requestId, now());
|
||||
if (planId) await operationLedger.start({ planId, nodeId, operation });
|
||||
if (operationKind === "readiness-probe") connection.latestReadinessProbe = operation;
|
||||
else connection.latestOperation = operation;
|
||||
const promise = new Promise((resolve) => {
|
||||
@@ -117,7 +151,7 @@ export function createHwpodNodeWsRegistry(options: any = {}) {
|
||||
finishPendingOperation({ nodeId, requestId, connection, operationKind, operation, timer, resolve }, "timed-out", now(), "hwpod_node_unavailable");
|
||||
resolve(blockedDispatch(plan, requestMeta, `hwpod-node ${nodeId} WebSocket dispatch timed out after ${timeoutMs}ms`));
|
||||
}, timeoutMs);
|
||||
const waiter = { nodeId, requestId, connection, operationKind, operation, timer, resolve } satisfies PendingDispatch;
|
||||
const waiter = { planId, nodeId, requestId, connection, operationKind, operation, timer, resolve } satisfies PendingDispatch;
|
||||
pending.set(requestId, waiter);
|
||||
try {
|
||||
sendJson(connection, { type: "hwpod-node-ops", requestId, plan, requestMeta });
|
||||
@@ -144,15 +178,11 @@ export function createHwpodNodeWsRegistry(options: any = {}) {
|
||||
return promise;
|
||||
}
|
||||
|
||||
function lookup(planId: string) {
|
||||
async function lookup(planId: string) {
|
||||
const authoritative = operations.get(safeText(planId));
|
||||
if (!authoritative) return null;
|
||||
return {
|
||||
planId: authoritative.planId,
|
||||
nodeId: authoritative.nodeId,
|
||||
operation: authoritative.operation,
|
||||
result: authoritative.result
|
||||
};
|
||||
if (authoritative) return { planId: authoritative.planId, nodeId: authoritative.nodeId, operation: authoritative.operation, result: authoritative.result };
|
||||
const durable = await operationLedger.get(safeText(planId));
|
||||
return durable ? { planId: durable.planId, nodeId: durable.nodeId, operation: durable.operation, result: durable.result } : null;
|
||||
}
|
||||
|
||||
function describe() {
|
||||
@@ -217,7 +247,7 @@ export function createHwpodNodeWsRegistry(options: any = {}) {
|
||||
const previous = connections.get(nodeId);
|
||||
if (previous && previous !== connection) previous.socket.close();
|
||||
connections.set(nodeId, connection);
|
||||
sendJson(connection, { type: "ack", requestId: "register", ok: true, message: "registered", nodeId });
|
||||
sendJson(connection, { type: "ack", requestId: "register", ok: true, message: "registered", nodeId, operationRetentionSeconds: Math.floor(operationLedger.retentionMs / 1000) });
|
||||
return;
|
||||
}
|
||||
if (messageType === "heartbeat") {
|
||||
@@ -233,10 +263,15 @@ export function createHwpodNodeWsRegistry(options: any = {}) {
|
||||
sendJson(connection, { type: "ack", requestId: safeText(message.requestId) || "diagnostic", ok: true, message: "diagnostic recorded" });
|
||||
return;
|
||||
}
|
||||
if (messageType === "hwpod-node-ops-result") completeDispatch(message);
|
||||
if (messageType === "hwpod-node-ops-accepted") {
|
||||
const waiter = pending.get(safeText(message.requestId));
|
||||
if (waiter?.planId) void operationLedger.accept(waiter.planId);
|
||||
return;
|
||||
}
|
||||
if (messageType === "hwpod-node-ops-result") void completeDispatch(message);
|
||||
}
|
||||
|
||||
function completeDispatch(message: any) {
|
||||
async function completeDispatch(message: any) {
|
||||
const requestId = safeText(message.requestId);
|
||||
const waiter = requestId ? pending.get(requestId) : null;
|
||||
if (!waiter) return;
|
||||
@@ -251,6 +286,8 @@ export function createHwpodNodeWsRegistry(options: any = {}) {
|
||||
now(),
|
||||
safeText(result.blocker?.code) || null
|
||||
);
|
||||
if (waiter.planId) await operationLedger.complete(waiter.planId, result);
|
||||
if (waiter.planId) operations.delete(waiter.planId);
|
||||
waiter.resolve(result);
|
||||
}
|
||||
|
||||
|
||||
@@ -0,0 +1,108 @@
|
||||
type HwpodOperationLedgerRecord = {
|
||||
planId: string;
|
||||
nodeId: string;
|
||||
operation: Record<string, unknown>;
|
||||
status: "started" | "accepted" | "completed";
|
||||
result: any | null;
|
||||
updatedAt: string;
|
||||
};
|
||||
|
||||
export function createHwpodOperationLedger(options: any = {}) {
|
||||
const runtimeStore = options.runtimeStore;
|
||||
const now = options.now ?? (() => new Date().toISOString());
|
||||
const retentionMs = positiveInteger(options.retentionMs, 0);
|
||||
if (!runtimeStore || typeof runtimeStore.query !== "function") return createMemoryHwpodOperationLedger({ now, retentionMs });
|
||||
|
||||
async function get(planId: string) {
|
||||
const result = await runtimeStore.query(
|
||||
"SELECT operation_json FROM hardware_operations WHERE id = $1 AND operation_type = 'hwpod.node-ops' LIMIT 1",
|
||||
[planId]
|
||||
);
|
||||
return result.rows?.[0] ? parseRecord(result.rows[0].operation_json) : null;
|
||||
}
|
||||
|
||||
async function write(record: HwpodOperationLedgerRecord) {
|
||||
await cleanup();
|
||||
await runtimeStore.query(
|
||||
`INSERT INTO hardware_operations (id, project_id, requested_by, operation_type, operation_json, status, requested_at, updated_at)
|
||||
VALUES ($1, NULL, 'hwlab-cloud-api', 'hwpod.node-ops', $2, $3, $4, $4)
|
||||
ON CONFLICT (id) DO UPDATE SET operation_json = EXCLUDED.operation_json, status = EXCLUDED.status, updated_at = EXCLUDED.updated_at`,
|
||||
[record.planId, JSON.stringify(record), record.status, record.updatedAt]
|
||||
);
|
||||
return record;
|
||||
}
|
||||
|
||||
async function start(input: Omit<HwpodOperationLedgerRecord, "status" | "result" | "updatedAt">) {
|
||||
const existing = await get(input.planId);
|
||||
return existing ?? write({ ...input, status: "started", result: null, updatedAt: now() });
|
||||
}
|
||||
|
||||
async function accept(planId: string) {
|
||||
const existing = await get(planId);
|
||||
return existing ? write({ ...existing, status: "accepted", updatedAt: now() }) : null;
|
||||
}
|
||||
|
||||
async function complete(planId: string, result: any) {
|
||||
const existing = await get(planId);
|
||||
return existing ? write({ ...existing, status: "completed", result, updatedAt: now() }) : null;
|
||||
}
|
||||
|
||||
async function cleanup() {
|
||||
if (!retentionMs) return;
|
||||
await runtimeStore.query(
|
||||
"DELETE FROM hardware_operations WHERE operation_type = 'hwpod.node-ops' AND updated_at < $1",
|
||||
[new Date(Date.now() - retentionMs).toISOString()]
|
||||
);
|
||||
}
|
||||
|
||||
return { get, start, accept, complete, retentionMs };
|
||||
}
|
||||
|
||||
export function createMemoryHwpodOperationLedger(options: any = {}) {
|
||||
const now = options.now ?? (() => new Date().toISOString());
|
||||
const retentionMs = positiveInteger(options.retentionMs, 0);
|
||||
const records = options.records ?? new Map<string, HwpodOperationLedgerRecord>();
|
||||
function cleanup() {
|
||||
if (!retentionMs) return;
|
||||
const cutoff = Date.now() - retentionMs;
|
||||
for (const [planId, record] of records) if (Date.parse(record.updatedAt) < cutoff) records.delete(planId);
|
||||
}
|
||||
async function get(planId: string) { cleanup(); return records.get(planId) ?? null; }
|
||||
async function start(input: Omit<HwpodOperationLedgerRecord, "status" | "result" | "updatedAt">) {
|
||||
cleanup();
|
||||
const existing = records.get(input.planId);
|
||||
if (existing) return existing;
|
||||
const record = { ...input, status: "started", result: null, updatedAt: now() } satisfies HwpodOperationLedgerRecord;
|
||||
records.set(input.planId, record);
|
||||
return record;
|
||||
}
|
||||
async function accept(planId: string) {
|
||||
const existing = records.get(planId);
|
||||
if (!existing) return null;
|
||||
const record = { ...existing, status: "accepted", updatedAt: now() } satisfies HwpodOperationLedgerRecord;
|
||||
records.set(planId, record);
|
||||
return record;
|
||||
}
|
||||
async function complete(planId: string, result: any) {
|
||||
const existing = records.get(planId);
|
||||
if (!existing) return null;
|
||||
const record = { ...existing, status: "completed", result, updatedAt: now() } satisfies HwpodOperationLedgerRecord;
|
||||
records.set(planId, record);
|
||||
return record;
|
||||
}
|
||||
return { get, start, accept, complete, retentionMs, records };
|
||||
}
|
||||
|
||||
function parseRecord(value: unknown): HwpodOperationLedgerRecord | null {
|
||||
try {
|
||||
const parsed = typeof value === "string" ? JSON.parse(value) : value;
|
||||
return parsed && typeof parsed === "object" && !Array.isArray(parsed) ? parsed as HwpodOperationLedgerRecord : null;
|
||||
} catch {
|
||||
return null;
|
||||
}
|
||||
}
|
||||
|
||||
function positiveInteger(value: unknown, fallback: number) {
|
||||
const parsed = Number.parseInt(String(value ?? ""), 10);
|
||||
return Number.isFinite(parsed) && parsed > 0 ? parsed : fallback;
|
||||
}
|
||||
@@ -34,7 +34,7 @@ export async function handleHwpodNodeOpsHttp(request, response, options) {
|
||||
if (request.method === "GET") {
|
||||
const planId = new URL(request.url, "http://hwlab.local").searchParams.get("planId")?.trim() ?? "";
|
||||
if (planId) {
|
||||
const operation = options.hwpodNodeWsRegistry.lookup(planId);
|
||||
const operation = await options.hwpodNodeWsRegistry.lookup(planId);
|
||||
sendJson(response, operation ? 200 : 404, operation
|
||||
? { ok: true, status: operation.result ? "completed" : "running", contractVersion: HWPOD_NODE_OPS_CONTRACT_VERSION, ...operation }
|
||||
: { ok: false, status: "not-found", contractVersion: HWPOD_NODE_OPS_CONTRACT_VERSION, planId, error: { code: "hwpod_operation_not_found", message: `hwpod operation ${planId} was not found` } });
|
||||
|
||||
@@ -172,7 +172,7 @@ export function createCloudApiServer(options = {}) {
|
||||
staleMs: parsePositiveInteger(env.HWLAB_GATEWAY_DEMO_STALE_MS, 30000),
|
||||
dispatchTimeoutMs: parsePositiveInteger(env.HWLAB_GATEWAY_DISPATCH_TIMEOUT_MS, 120000)
|
||||
});
|
||||
const hwpodNodeWsRegistry = options.hwpodNodeWsRegistry || createHwpodNodeWsRegistry({ ...options, env });
|
||||
const hwpodNodeWsRegistry = options.hwpodNodeWsRegistry || createHwpodNodeWsRegistry({ ...options, env, runtimeStore });
|
||||
const accessController = options.accessController || createAccessController({ ...options, env, runtimeStore, gatewayRegistry, traceStore, codeAgentChatResults });
|
||||
if (typeof accessController.configureCodeAgentWorkspaceContext === "function") {
|
||||
accessController.configureCodeAgentWorkspaceContext({ env, fetchImpl: options.fetchImpl, traceStore, codeAgentChatResults });
|
||||
|
||||
@@ -158,12 +158,18 @@ async function idempotent(registry: HarnessRLRegistry, input: { runId: string; i
|
||||
const activityName = input.identity.split(":").at(-2) ?? "activity";
|
||||
const existing = await registry.getActivityResult(input.runId, activityName, input.identity);
|
||||
if ((existing?.output as any)?.state === "completed") return (existing!.output as any).result;
|
||||
if ((existing?.output as any)?.state === "accepted" && recover) {
|
||||
const output = await recover();
|
||||
if (["started", "accepted"].includes((existing?.output as any)?.state) && recover) {
|
||||
let output: Record<string, unknown>;
|
||||
try {
|
||||
output = await recover();
|
||||
} catch (error: any) {
|
||||
if (!authorityNotFound(error)) throw error;
|
||||
output = await execute();
|
||||
}
|
||||
const saved = await registry.saveActivityResult({ runId: input.runId, activityName, identity: input.identity, output: { state: "completed", authorityIdentity: input.identity, result: output } });
|
||||
return activityOutput(saved);
|
||||
}
|
||||
if (!existing) await registry.saveActivityResult({ runId: input.runId, activityName, identity: input.identity, output: { state: "accepted", authorityIdentity: input.identity } });
|
||||
if (!existing) await registry.saveActivityResult({ runId: input.runId, activityName, identity: input.identity, output: { state: "started", authorityIdentity: input.identity } });
|
||||
const output = await execute();
|
||||
const saved = await registry.saveActivityResult({ runId: input.runId, activityName, identity: input.identity, output: { state: "completed", authorityIdentity: input.identity, result: output } });
|
||||
return activityOutput(saved);
|
||||
@@ -189,6 +195,11 @@ function activityOutput(result: { output: Record<string, unknown> }) {
|
||||
return (result.output as any)?.state === "completed" ? (result.output as any).result : result.output;
|
||||
}
|
||||
|
||||
function authorityNotFound(error: any) {
|
||||
const code = String(error?.code ?? "");
|
||||
return code === "hwpod_operation_not_found" || code.endsWith("_not_found");
|
||||
}
|
||||
|
||||
async function manifestRefs(summary: any) {
|
||||
const manifestPath = String(summary?.artifactManifestPath ?? "");
|
||||
if (!manifestPath) return { path: null, sha256: null, aggregate: null };
|
||||
|
||||
@@ -58,6 +58,32 @@ test("worker restart reuses idempotent prepare/build/collect activity results",
|
||||
assert.equal(registry.activityResults.size, 6);
|
||||
});
|
||||
|
||||
test("worker uses configured cloud-api authority and resubmits the same identity after typed not-found", async () => {
|
||||
const fixture = await softwareSmokeFixture();
|
||||
const registry = new MemoryRegistry();
|
||||
const runId = "authority-retry";
|
||||
const identity = `${runId}:agent:v1`;
|
||||
await registry.createRun({ runId, caseId: fixture.caseId, workflowId: `workflow-${runId}`, caseRepo: fixture.root, runDir: path.join(fixture.stateRoot, "runs", runId), runtimeApiUrl: "https://public.example.test" });
|
||||
await registry.saveActivityResult({ runId, activityName: "prepare", identity: `${runId}:prepare:v1`, output: { state: "completed", result: { mode: "hardware", prepared: { run: { runId } } } } });
|
||||
await registry.saveActivityResult({ runId, activityName: "agent", identity, output: { state: "started", authorityIdentity: identity } });
|
||||
let submitCount = 0;
|
||||
let recoverCount = 0;
|
||||
const activities = createHarnessRLActivities({
|
||||
registry,
|
||||
env: { HWLAB_CASERUN_INTERNAL_API_URL: "http://hwlab-cloud-api.hwlab-v03.svc.cluster.local:6667" },
|
||||
stages: {
|
||||
agent: async (context: any, run: any) => { submitCount += 1; assert.equal(context.parsed.apiUrl, "http://hwlab-cloud-api.hwlab-v03.svc.cluster.local:6667"); assert.equal(context.parsed.operationIdentity, identity); return { run, agent: { traceId: "trace-authority-retry" } }; },
|
||||
recoverAgent: async () => { recoverCount += 1; throw Object.assign(new Error("not found"), { code: "agentrun_task_not_found" }); }
|
||||
}
|
||||
});
|
||||
const result = await activities.agent({ runId, identity });
|
||||
assert.equal(result.agent.traceId, "trace-authority-retry");
|
||||
assert.equal(recoverCount, 1);
|
||||
assert.equal(submitCount, 1);
|
||||
const deploy = await readFile(path.join(process.cwd(), "deploy", "deploy.yaml"), "utf8");
|
||||
assert.match(deploy, /HWLAB_CASERUN_INTERNAL_API_URL: http:\/\/hwlab-cloud-api\.hwlab-v03\.svc\.cluster\.local:6667/u);
|
||||
});
|
||||
|
||||
test("AgentRun and HWPOD crash gaps reuse stable authoritative operation identities", async () => {
|
||||
const fixture = await hardwareFixture();
|
||||
const registry = new MemoryRegistry();
|
||||
|
||||
+38
-1
@@ -11,6 +11,7 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import base64
|
||||
import calendar
|
||||
import ctypes
|
||||
import ctypes.wintypes as wt
|
||||
import hashlib
|
||||
@@ -61,6 +62,7 @@ LOG_DIR = CONFIG_DIR / "logs"
|
||||
UPDATE_DIR = CONFIG_DIR / "update"
|
||||
CONFIG_PATH = CONFIG_DIR / "config.json"
|
||||
STATE_PATH = CONFIG_DIR / "state.json"
|
||||
OPERATION_LEDGER_PATH = CONFIG_DIR / "hwpod-operations.json"
|
||||
LOG_PATH = LOG_DIR / "hwlab-node.log"
|
||||
|
||||
DEFAULT_CONFIG = {
|
||||
@@ -832,6 +834,7 @@ class WebSocketNodeClient:
|
||||
self.manual_stop = False
|
||||
self.connected = False
|
||||
self.diagnostic_cursor = 0
|
||||
self.operation_retention_seconds = 0
|
||||
|
||||
def start(self) -> None:
|
||||
if self.thread and self.thread.is_alive():
|
||||
@@ -972,18 +975,52 @@ class WebSocketNodeClient:
|
||||
message = json.loads(text)
|
||||
if message.get("type") == "ack" and message.get("requestId") == "register":
|
||||
self.connected = message.get("ok") is True
|
||||
self.operation_retention_seconds = safe_int(message.get("operationRetentionSeconds"), 0)
|
||||
self.bus.emit("connection", "connected" if self.connected else "error")
|
||||
self.bus.emit("log", f"注册结果: ok={message.get('ok')} message={message.get('message')}")
|
||||
return
|
||||
if message.get("type") != "hwpod-node-ops":
|
||||
return
|
||||
request_id = str(message.get("requestId") or "req_unknown")
|
||||
plan = message.get("plan") or {}
|
||||
plan_id = str(plan.get("planId") or request_id)
|
||||
ledger = self._operation_ledger()
|
||||
existing = ledger.get(plan_id)
|
||||
self._send_json({"type": "hwpod-node-ops-accepted", "nodeId": self.executor.node_id, "requestId": request_id, "planId": plan_id, "observedAt": utc_now()})
|
||||
if isinstance(existing, dict) and existing.get("status") == "completed" and isinstance(existing.get("result"), dict):
|
||||
self._send_json({"type": "hwpod-node-ops-result", "nodeId": self.executor.node_id, "requestId": request_id, "result": existing["result"], "observedAt": utc_now()})
|
||||
return
|
||||
ledger[plan_id] = {"status": "accepted", "acceptedAt": utc_now(), "result": None}
|
||||
self._save_operation_ledger(ledger)
|
||||
try:
|
||||
result = self.executor.execute(message.get("plan") or {})
|
||||
result = self.executor.execute(plan)
|
||||
except Exception as exc:
|
||||
result = {"ok": False, "status": "failed", "results": [], "blocker": {"code": "hwpod_node_ws_run_failed", "layer": "hwpod-node", "retryable": True, "summary": str(exc)}}
|
||||
ledger[plan_id] = {"status": "completed", "acceptedAt": ledger[plan_id]["acceptedAt"], "completedAt": utc_now(), "result": result}
|
||||
self._save_operation_ledger(ledger)
|
||||
self._send_json({"type": "hwpod-node-ops-result", "nodeId": self.executor.node_id, "requestId": request_id, "result": result, "observedAt": utc_now()})
|
||||
|
||||
def _operation_ledger(self) -> dict:
|
||||
try:
|
||||
value = json.loads(OPERATION_LEDGER_PATH.read_text(encoding="utf-8"))
|
||||
ledger = value if isinstance(value, dict) else {}
|
||||
except Exception:
|
||||
ledger = {}
|
||||
if self.operation_retention_seconds > 0:
|
||||
cutoff = time.time() - self.operation_retention_seconds
|
||||
ledger = {key: value for key, value in ledger.items() if not isinstance(value, dict) or self._operation_timestamp(value) >= cutoff}
|
||||
return ledger
|
||||
|
||||
def _save_operation_ledger(self, ledger: dict) -> None:
|
||||
atomic_write_json(OPERATION_LEDGER_PATH, ledger)
|
||||
|
||||
def _operation_timestamp(self, value: dict) -> float:
|
||||
timestamp = str(value.get("completedAt") or value.get("acceptedAt") or "")
|
||||
try:
|
||||
return calendar.timegm(time.strptime(timestamp, "%Y-%m-%dT%H:%M:%SZ"))
|
||||
except Exception:
|
||||
return time.time()
|
||||
|
||||
def _flush_diagnostics(self) -> None:
|
||||
for event in DIAGNOSTICS.since(self.diagnostic_cursor, 20):
|
||||
self._send_json({"type": "hwpod-node-diagnostic", "nodeId": self.executor.node_id, "diagnostic": event, "observedAt": utc_now()})
|
||||
|
||||
@@ -86,6 +86,9 @@ export function connectHwpodNodeWs(config: any, options: any = {}) {
|
||||
let closed = false;
|
||||
let attempt = 0;
|
||||
let readyResolve: ((value: unknown) => void) | null = null;
|
||||
const operations = new Map<string, Promise<any>>();
|
||||
const operationCompletedAt = new Map<string, number>();
|
||||
let operationRetentionMs = 0;
|
||||
const ready = new Promise((resolve) => { readyResolve = resolve; });
|
||||
|
||||
function connect() {
|
||||
@@ -129,19 +132,24 @@ export function connectHwpodNodeWs(config: any, options: any = {}) {
|
||||
const textValue = typeof raw === "string" ? raw : await new Response(raw).text();
|
||||
const message = JSON.parse(textValue);
|
||||
if (message?.type === "ack" && message.requestId === "register" && readyResolve) {
|
||||
operationRetentionMs = Math.max(0, (numberValue(message.operationRetentionSeconds) ?? 0) * 1000);
|
||||
logWsEvent("registered", { ok: message.ok === true, message: text(message.message) || null });
|
||||
readyResolve(message);
|
||||
readyResolve = null;
|
||||
return;
|
||||
}
|
||||
if (message?.type !== "hwpod-node-ops") return;
|
||||
pruneOperations();
|
||||
const requestId = text(message.requestId) || "req_unknown";
|
||||
let resultPayload;
|
||||
try {
|
||||
resultPayload = await executePlan(message.plan);
|
||||
} catch (error) {
|
||||
resultPayload = failure("hwpod-node.ws.run", error);
|
||||
const planId = text(message.plan?.planId) || requestId;
|
||||
sendJson({ type: "hwpod-node-ops-accepted", nodeId, requestId, planId, observedAt: now() });
|
||||
let operation = operations.get(planId);
|
||||
if (!operation) {
|
||||
operation = Promise.resolve().then(() => executePlan(message.plan)).catch((error) => failure("hwpod-node.ws.run", error));
|
||||
operations.set(planId, operation);
|
||||
void operation.finally(() => operationCompletedAt.set(planId, Date.now()));
|
||||
}
|
||||
const resultPayload = await operation;
|
||||
sendJson({ type: "hwpod-node-ops-result", nodeId, requestId, result: resultPayload, observedAt: now() });
|
||||
}
|
||||
|
||||
@@ -149,6 +157,16 @@ export function connectHwpodNodeWs(config: any, options: any = {}) {
|
||||
sendJson({ type: "heartbeat", nodeId, at: now() });
|
||||
}
|
||||
|
||||
function pruneOperations() {
|
||||
if (!operationRetentionMs) return;
|
||||
const cutoff = Date.now() - operationRetentionMs;
|
||||
for (const [planId, completedAt] of operationCompletedAt) {
|
||||
if (completedAt >= cutoff) continue;
|
||||
operationCompletedAt.delete(planId);
|
||||
operations.delete(planId);
|
||||
}
|
||||
}
|
||||
|
||||
function sendJson(value: any) {
|
||||
if (!socket || socket.readyState !== WebSocket.OPEN) return false;
|
||||
socket.send(JSON.stringify(value));
|
||||
|
||||
Reference in New Issue
Block a user