From 9715992cd4f85a2f534e0e7c3fbd3cc91d1f1769 Mon Sep 17 00:00:00 2001 From: root Date: Fri, 17 Jul 2026 04:48:32 +0200 Subject: [PATCH] =?UTF-8?q?fix:=20=E6=81=A2=E5=A4=8D=20HarnessRL=20HWPOD?= =?UTF-8?q?=20=E9=87=8D=E5=90=AF=E5=B9=82=E7=AD=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- deploy/deploy.yaml | 4 + internal/cloud/bun-server.ts | 6 +- internal/cloud/hwpod-node-ops.test.ts | 44 ++++++++- internal/cloud/hwpod-node-ws-registry.ts | 69 +++++++++++---- internal/cloud/hwpod-operation-ledger.ts | 108 +++++++++++++++++++++++ internal/cloud/server-hwpod-http.ts | 2 +- internal/cloud/server.ts | 2 +- internal/harnessrl/activities.ts | 17 +++- internal/harnessrl/harnessrl.test.ts | 26 ++++++ tools/hwlab-node.py | 39 +++++++- tools/src/hwpod-node-lib.ts | 28 ++++-- 11 files changed, 315 insertions(+), 30 deletions(-) create mode 100644 internal/cloud/hwpod-operation-ledger.ts diff --git a/deploy/deploy.yaml b/deploy/deploy.yaml index cf0d5909..eabe225c 100644 --- a/deploy/deploy.yaml +++ b/deploy/deploy.yaml @@ -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 diff --git a/internal/cloud/bun-server.ts b/internal/cloud/bun-server.ts index 78430ebc..963ce82c 100644 --- a/internal/cloud/bun-server.ts +++ b/internal/cloud/bun-server.ts @@ -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) { diff --git a/internal/cloud/hwpod-node-ops.test.ts b/internal/cloud/hwpod-node-ops.test.ts index 722a40bc..56185612 100644 --- a/internal/cloud/hwpod-node-ops.test.ts +++ b/internal/cloud/hwpod-node-ops.test.ts @@ -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() {} }; diff --git a/internal/cloud/hwpod-node-ws-registry.ts b/internal/cloud/hwpod-node-ws-registry.ts index 23818d75..b293e269 100644 --- a/internal/cloud/hwpod-node-ws-registry.ts +++ b/internal/cloud/hwpod-node-ws-registry.ts @@ -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(); const pending = new Map(); const operations = new Map(); + const admissions = new Map }>(); const socketConnections = new WeakMap(); 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 || ""} is not connected through outbound WebSocket`)); + return blockedDispatch(plan, requestMeta, `hwpod-node ${nodeId || ""} 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); } diff --git a/internal/cloud/hwpod-operation-ledger.ts b/internal/cloud/hwpod-operation-ledger.ts new file mode 100644 index 00000000..d7f313d6 --- /dev/null +++ b/internal/cloud/hwpod-operation-ledger.ts @@ -0,0 +1,108 @@ +type HwpodOperationLedgerRecord = { + planId: string; + nodeId: string; + operation: Record; + 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) { + 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(); + 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) { + 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; +} diff --git a/internal/cloud/server-hwpod-http.ts b/internal/cloud/server-hwpod-http.ts index be53cb35..ba035623 100644 --- a/internal/cloud/server-hwpod-http.ts +++ b/internal/cloud/server-hwpod-http.ts @@ -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` } }); diff --git a/internal/cloud/server.ts b/internal/cloud/server.ts index 7fa7576d..2395f88f 100644 --- a/internal/cloud/server.ts +++ b/internal/cloud/server.ts @@ -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 }); diff --git a/internal/harnessrl/activities.ts b/internal/harnessrl/activities.ts index b2770bc9..2aa7413d 100644 --- a/internal/harnessrl/activities.ts +++ b/internal/harnessrl/activities.ts @@ -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; + 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 }) { 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 }; diff --git a/internal/harnessrl/harnessrl.test.ts b/internal/harnessrl/harnessrl.test.ts index 38baa114..1f5c1b63 100644 --- a/internal/harnessrl/harnessrl.test.ts +++ b/internal/harnessrl/harnessrl.test.ts @@ -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(); diff --git a/tools/hwlab-node.py b/tools/hwlab-node.py index b23a7dcc..a786b51c 100644 --- a/tools/hwlab-node.py +++ b/tools/hwlab-node.py @@ -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()}) diff --git a/tools/src/hwpod-node-lib.ts b/tools/src/hwpod-node-lib.ts index b6d5063d..fd996c1d 100644 --- a/tools/src/hwpod-node-lib.ts +++ b/tools/src/hwpod-node-lib.ts @@ -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>(); + const operationCompletedAt = new Map(); + 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));