From abb1a4344bfae0fa7b7afce033e2cca5155f8d15 Mon Sep 17 00:00:00 2001 From: aaaamoooo Date: Tue, 20 Oct 2026 07:46:33 +0000 Subject: [PATCH] =?UTF-8?q?feat:=20heartbeat=20reconciliation=20=E2=80=94?= =?UTF-8?q?=20panel=20re-dispatches=20resources=20a=20node=20lost?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Agents (Go + Node) now report the resource IDs they actually hold in each heartbeat. When an expected resource is missing locally (e.g. the node's /etc/nexusgate state was wiped and only the agent restarted, so no re-enrollment happens), the panel re-queues applies for just the affected nodes via reconcileFromHeartbeat: gated on no queued/running jobs for the server (in-flight jobs are never failed by this path) and throttled to one trigger per server per 120s, audited as system:heartbeat. reconcileServer gains optional onlyNodeIds / failInFlight parameters; the enrollment call is unchanged. Heartbeats without a resources array (old agents) behave exactly as before. --- agent-go/health.go | 35 +++-- agent/agent.js | 6 +- lib/orchestrator.js | 118 ++++++++++++----- server.js | 8 ++ test/heartbeat-reconcile.test.js | 221 +++++++++++++++++++++++++++++++ 5 files changed, 342 insertions(+), 46 deletions(-) create mode 100644 test/heartbeat-reconcile.test.js diff --git a/agent-go/health.go b/agent-go/health.go index e37f772..38d6f0f 100644 --- a/agent-go/health.go +++ b/agent-go/health.go @@ -201,13 +201,9 @@ type engineReport struct { SingBoxStatus string `json:"singBoxStatus"` } -// engineHealth mirrors agent.js engineHealth(). Callers must hold stateMu. -// A readResources error propagates (JS throws, so the heartbeat is skipped). -func engineHealthLocked() (engineReport, error) { - resources, err := readResources() - if err != nil { - return engineReport{}, err - } +// engineHealth mirrors agent.js engineHealth(). Callers must hold stateMu +// and pass the resources they read under the same lock. +func engineHealthLocked(resources []map[string]interface{}) engineReport { installed := fileExists(cfg.SingBoxBin) && fileExists(cfg.SingBoxStatsBin) service := serviceHealth("nexusgate-sing-box") result := engineReport{ @@ -236,20 +232,33 @@ func engineHealthLocked() (engineReport, error) { result.Detail = "Hysteria 2 UDP 未监听：" + strings.Join(parts, ", ") } } - return result, nil + return result } func heartbeat() error { stateMu.Lock() - engine, err := engineHealthLocked() - stateMu.Unlock() + // A readResources error skips the heartbeat entirely (JS throws). + resources, err := readResources() if err != nil { + stateMu.Unlock() return err } + engine := engineHealthLocked(resources) + stateMu.Unlock() + // The panel reconciles against this list: any expected resource the node + // no longer holds (e.g. local state was wiped) is re-dispatched + // automatically. An empty list is meaningful, so keep it non-nil. + resourceIDs := []string{} + for _, res := range resources { + if s, ok := res["id"].(string); ok && s != "" { + resourceIDs = append(resourceIDs, s) + } + } body := map[string]interface{}{ - "version": version, - "system": systemInfo(), - "engine": engine, + "version": version, + "system": systemInfo(), + "engine": engine, + "resources": resourceIDs, } if _, err := requestJSON("/api/agent/heartbeat", body); err != nil { return err diff --git a/agent/agent.js b/agent/agent.js index 3c78333..2e53982 100644 --- a/agent/agent.js +++ b/agent/agent.js @@ -339,7 +339,11 @@ function serviceHealth(service) { } async function heartbeat() { - await request('/api/agent/heartbeat', { method: 'POST', body: JSON.stringify({ version: VERSION, system: systemInfo(), engine: engineHealth() }) }); + const engine = engineHealth(); + // Report the resources actually held locally; the panel re-dispatches any + // expected resource missing from this list (e.g. after a state wipe). + const resources = readResources().map((resource) => resource && resource.id).filter((value) => typeof value === 'string' && value); + await request('/api/agent/heartbeat', { method: 'POST', body: JSON.stringify({ version: VERSION, system: systemInfo(), engine, resources }) }); try { fs.writeFileSync('/etc/nexusgate/last-heartbeat.json', `${JSON.stringify({ at:new Date().toISOString() })}\n`, { mode:0o600 }); } catch (error) { log('Heartbeat reached controller but readiness file failed', error.message); } } diff --git a/lib/orchestrator.js b/lib/orchestrator.js index 526de7c..3a5d4c4 100644 --- a/lib/orchestrator.js +++ b/lib/orchestrator.js @@ -12,6 +12,9 @@ const { const nowIso = () => new Date().toISOString(); const id = (prefix) => `${prefix}_${crypto.randomUUID()}`; +// Minimum gap between two heartbeat-triggered reconciliations of one server. +const HEARTBEAT_RECONCILE_INTERVAL_MS = 120000; + function assertFound(value, label) { if (!value) { const error = new Error(`${label}不存在`); @@ -270,8 +273,53 @@ function validateNodeDeploy(data, node) { if (!direct && nodePortInUse(data, node.exitServerId, node.exitPort, node.id)) { const error = new Error(`出口端口 ${node.exitPort} 已被该服务器上的其他节点占用`); error.statusCode = 409; throw error; } } +// Re-queue apply/delete jobs so the agent's local resources converge to the +// panel's desired state. onlyNodeIds restricts the pass to those nodes (used +// by heartbeat reconciliation to heal just the nodes with missing resources); +// failInFlight marks queued/running jobs as superseded (enrollment semantics: +// a re-registering agent abandons its old leases). Heartbeat reconciliation +// gates on "no in-flight jobs" instead and passes failInFlight: false. +async function reconcileServerData(data, serverId, actor, { onlyNodeIds = null, failInFlight = true } = {}) { + const server = assertFound(data.servers.find((item) => item.id === serverId), '服务器'); + if (failInFlight) { + for (const job of data.jobs.filter((item) => item.serverId === serverId && ['queued', 'running'].includes(item.status))) { + job.status = 'failed'; job.error = 'Agent 重新注册，旧任务已由对账任务替代'; job.updatedAt = nowIso(); + } + } + let count = 0; + for (const node of data.nodes.filter((n) => n.serverId === serverId || n.exitServerId === serverId)) { + if (onlyNodeIds && !onlyNodeIds.has(node.id)) continue; + const isEntry = node.serverId === serverId; + const isExit = node.exitServerId === serverId && forwardNode(node); + if (!isEntry && !isExit) continue; + if (node.status === 'draft') continue; + if (node.status === 'removing') { + if (isEntry) queueJob(data, serverId, node.id, 'delete_resource', { resourceId: node.resourceId, kind: 'node' }); + if (isExit && node.exitResourceId) queueJob(data, serverId, node.id, 'delete_resource', { resourceId: node.exitResourceId, kind: 'node-exit' }); + } else { + const clients = activeNodeClients(data, node); + if (!clients.length) continue; + if (isEntry) { + const payload = { resource: await resourceForNode(data, node, clients), kind: 'node' }; + const keyMaterial = keyMaterialForNode(data, node); + if (keyMaterial) payload.keyMaterial = keyMaterial; + queueJob(data, serverId, node.id, 'apply_resource', payload); + } + if (isExit) queueJob(data, serverId, node.id, 'apply_resource', { resource: await resourceForNodeExit(data, node), kind: 'node-exit' }); + node.status = 'deploying'; + } + node.updatedAt = nowIso(); count += 1; + } + audit(data, actor, 'reconcile_server', server.id, { nodes: count }); + return count; +} + class Orchestrator { - constructor(store) { this.store = store; } + constructor(store) { + this.store = store; + // serverId -> epoch ms of the last heartbeat-triggered reconciliation. + this.heartbeatReconcileAt = new Map(); + } async deployNode(nodeId, actor = 'admin') { return this.store.transaction(async (data) => { @@ -360,38 +408,44 @@ class Orchestrator { }); } - async reconcileServer(serverId, actor = 'system') { - return this.store.transaction(async (data) => { - const server = assertFound(data.servers.find((item) => item.id === serverId), '服务器'); - for (const job of data.jobs.filter((item) => item.serverId === serverId && ['queued', 'running'].includes(item.status))) { - job.status = 'failed'; job.error = 'Agent 重新注册，旧任务已由对账任务替代'; job.updatedAt = nowIso(); + async reconcileServer(serverId, actor = 'system', options = {}) { + return this.store.transaction(async (data) => reconcileServerData(data, serverId, actor, options)); + } + + // Heartbeat-driven reconciliation. Newer agents report the resource IDs + // they actually hold; when a resource the panel expects is missing locally + // (e.g. the node's /etc/nexusgate state was wiped and only the agent was + // restarted, so no re-enrollment ever happens), re-queue applies for just + // the affected nodes. Old agents never send `resources`, and callers only + // invoke this for array reports, so their behavior is unchanged. Resources + // the agent holds but the panel no longer expects are NOT cleaned up here. + // Gates: never run while the server has queued/running jobs (an in-flight + // deploy explains the gap), and at most one trigger per server per + // HEARTBEAT_RECONCILE_INTERVAL_MS so a node that cannot heal does not spin. + async reconcileFromHeartbeat(serverId, reportedResourceIds) { + if (!Array.isArray(reportedResourceIds)) return 0; + const last = this.heartbeatReconcileAt.get(serverId) || 0; + if (Date.now() - last < HEARTBEAT_RECONCILE_INTERVAL_MS) return 0; + const reported = new Set(reportedResourceIds.slice(0, 1000).filter((value) => typeof value === 'string' && value)); + let triggered = false; + const count = await this.store.transaction(async (data) => { + const server = data.servers.find((item) => item.id === serverId); + if (!server) return 0; + const missingNodeIds = new Set(); + for (const node of data.nodes) { + if (node.status === 'draft' || node.status === 'removing') continue; + if (!activeNodeClients(data, node).length) continue; + if (node.serverId === serverId && node.resourceId && !reported.has(node.resourceId)) missingNodeIds.add(node.id); + if (node.exitServerId === serverId && forwardNode(node) && node.exitResourceId && !reported.has(node.exitResourceId)) missingNodeIds.add(node.id); } - let count = 0; - for (const node of data.nodes.filter((n) => n.serverId === serverId || n.exitServerId === serverId)) { - const isEntry = node.serverId === serverId; - const isExit = node.exitServerId === serverId && forwardNode(node); - if (!isEntry && !isExit) continue; - if (node.status === 'draft') continue; - if (node.status === 'removing') { - if (isEntry) queueJob(data, serverId, node.id, 'delete_resource', { resourceId: node.resourceId, kind: 'node' }); - if (isExit && node.exitResourceId) queueJob(data, serverId, node.id, 'delete_resource', { resourceId: node.exitResourceId, kind: 'node-exit' }); - } else { - const clients = activeNodeClients(data, node); - if (!clients.length) continue; - if (isEntry) { - const payload = { resource: await resourceForNode(data, node, clients), kind: 'node' }; - const keyMaterial = keyMaterialForNode(data, node); - if (keyMaterial) payload.keyMaterial = keyMaterial; - queueJob(data, serverId, node.id, 'apply_resource', payload); - } - if (isExit) queueJob(data, serverId, node.id, 'apply_resource', { resource: await resourceForNodeExit(data, node), kind: 'node-exit' }); - node.status = 'deploying'; - } - node.updatedAt = nowIso(); count += 1; - } - audit(data, actor, 'reconcile_server', server.id, { nodes: count }); - return count; + if (!missingNodeIds.size) return 0; + const inFlight = data.jobs.some((job) => job.serverId === serverId && ['queued', 'running'].includes(job.status)); + if (inFlight) return 0; + triggered = true; + return reconcileServerData(data, serverId, 'system:heartbeat', { onlyNodeIds: missingNodeIds, failInFlight: false }); }); + if (triggered) this.heartbeatReconcileAt.set(serverId, Date.now()); + return count; } async completeJob(jobId, success, result = {}, errorMessage = null) { @@ -487,4 +541,4 @@ class Orchestrator { module.exports = { Orchestrator, allocatePort, selectPort, nodePortInUse, queueJob, audit, id, nowIso, cancelQueuedDeletes, resourceForNode, resourceForNodeExit, rebuildNode, applyDelta, forwardNode, activeNodeClients, - suspendCustomerResources, resumeCustomerResources }; + suspendCustomerResources, resumeCustomerResources, HEARTBEAT_RECONCILE_INTERVAL_MS }; diff --git a/server.js b/server.js index fb28142..ccdea2e 100644 --- a/server.js +++ b/server.js @@ -308,6 +308,14 @@ async function handleAgent(req, res, pathname) { : server.engine || null; } }); + // Agents that report their locally held resource IDs get automatic + // re-dispatch of anything the panel expects but the node lost (e.g. a + // wiped /etc/nexusgate with no re-enrollment). Reports without the field + // (older agents) skip this entirely. + if (Array.isArray(body.resources)) { + try { await orchestrator.reconcileFromHeartbeat(agent.serverId, body.resources); } + catch (error) { console.error('Heartbeat reconciliation failed:', error); } + } sendJson(res, 200, { ok: true, serverTime: nowIso() }); return true; } diff --git a/test/heartbeat-reconcile.test.js b/test/heartbeat-reconcile.test.js new file mode 100644 index 0000000..17c277a --- /dev/null +++ b/test/heartbeat-reconcile.test.js @@ -0,0 +1,221 @@ +'use strict'; + +const test = require('node:test'); +const assert = require('node:assert/strict'); +const crypto = require('node:crypto'); +const fs = require('node:fs'); +const os = require('node:os'); +const path = require('node:path'); +const net = require('node:net'); +const { spawn } = require('node:child_process'); +const { Store } = require('../lib/store'); +const { Orchestrator } = require('../lib/orchestrator'); + +const B64URL_43 = /^[A-Za-z0-9_-]{43}$/; + +// One server, one customer, one active direct Reality node whose resource the +// panel believes is deployed on the agent. +async function fixture(t) { + const dir = fs.mkdtempSync(path.join(os.tmpdir(), 'nexusgate-hb-')); + t.after(() => fs.rmSync(dir, { recursive: true, force: true })); + const store = await new Store(path.join(dir, 'data.json')).init(); + const stamp = new Date().toISOString(); + await store.transaction((data) => { + data.servers.push({ id: 'srv_hb', name: '入口', publicAddress: '203.0.113.10', portRangeStart: 20000, portRangeEnd: 50000, status: 'online', lastSeenAt: stamp }); + data.customers.push({ id: 'cus_hb', name: '客户', status: 'active' }); + data.nodes.push({ + id: 'nod_hb', serverId: 'srv_hb', name: 'Reality 节点', protocol: 'vless-reality-vision', + topology: 'direct', networkMode: 'ipv4', port: 23456, resourceId: 'res_nod_hb_1', + status: 'active', entryReady: true, realityServerName: 'www.tesla.com', createdAt: stamp, updatedAt: stamp, + }); + data.nodeClients.push({ + id: 'ncl_hb', nodeId: 'nod_hb', customerId: 'cus_hb', status: 'active', + credentials: { clientId: crypto.randomUUID(), relayPassword: 'secret', shortId: 'abcd1234' }, + createdAt: stamp, updatedAt: stamp, + }); + }); + return { store, orchestrator: new Orchestrator(store) }; +} + +function applyJobs(store, nodeId) { + return store.data.jobs.filter((job) => job.action === 'apply_resource' && (!nodeId || job.nodeId === nodeId)); +} + +test('heartbeat reconcile requeues an apply with keyMaterial for a missing resource', async (t) => { + const { store, orchestrator } = await fixture(t); + const count = await orchestrator.reconcileFromHeartbeat('srv_hb', []); + assert.equal(count, 1); + const jobs = applyJobs(store, 'nod_hb'); + assert.equal(jobs.length, 1); + assert.equal(jobs[0].status, 'queued'); + assert.equal(jobs[0].serverId, 'srv_hb'); + assert.equal(jobs[0].payload.kind, 'node'); + const material = jobs[0].payload.keyMaterial; + assert.deepEqual(Object.keys(material), ['reality-nod_hb']); + assert.match(material['reality-nod_hb'].privateKey, B64URL_43); + assert.equal(store.data.keyStore['reality-nod_hb'].privateKey, material['reality-nod_hb'].privateKey); + assert.equal(store.data.nodes[0].status, 'deploying'); + assert.ok(store.data.activity.some((event) => event.action === 'reconcile_server' && event.actor === 'system:heartbeat')); +}); + +test('heartbeat reconcile is a no-op when the agent holds every expected resource', async (t) => { + const { store, orchestrator } = await fixture(t); + assert.equal(await orchestrator.reconcileFromHeartbeat('srv_hb', ['res_nod_hb_1']), 0); + assert.equal(store.data.jobs.length, 0); +}); + +test('heartbeat reconcile does not trigger or disturb jobs while a deploy is in flight', async (t) => { + const { store, orchestrator } = await fixture(t); + const stamp = new Date().toISOString(); + await store.transaction((data) => { + data.jobs.push({ id: 'job_busy', serverId: 'srv_hb', nodeId: 'nod_hb', action: 'apply_resource', + payload: { kind: 'node' }, status: 'queued', attempts: 0, createdAt: stamp, updatedAt: stamp }); + }); + assert.equal(await orchestrator.reconcileFromHeartbeat('srv_hb', []), 0); + assert.equal(store.data.jobs.length, 1); + // The in-flight job must NOT be failed the way enrollment reconcile fails it. + assert.equal(store.data.jobs[0].status, 'queued'); + assert.equal(store.data.activity.some((event) => event.action === 'reconcile_server'), false); +}); + +test('heartbeat reconcile is throttled per server', async (t) => { + const { store, orchestrator } = await fixture(t); + assert.equal(await orchestrator.reconcileFromHeartbeat('srv_hb', []), 1); + assert.equal(applyJobs(store, 'nod_hb').length, 1); + // Clear the in-flight gate so only the throttle can explain a no-op. + await store.transaction((data) => { data.jobs[0].status = 'completed'; }); + assert.equal(await orchestrator.reconcileFromHeartbeat('srv_hb', []), 0); + assert.equal(applyJobs(store, 'nod_hb').length, 1); +}); + +test('heartbeat reconcile ignores legacy reports and out-of-scope nodes', async (t) => { + const { store, orchestrator } = await fixture(t); + // Old agents: no resources field at all. + assert.equal(await orchestrator.reconcileFromHeartbeat('srv_hb', undefined), 0); + assert.equal(await orchestrator.reconcileFromHeartbeat('srv_hb', 'res_nod_hb_1'), 0); + const stamp = new Date().toISOString(); + await store.transaction((data) => { + // Draft node (never deployed), clientless node, and a removing node: + // none of their resources are "expected" on the agent. + data.nodes.push( + { id: 'nod_draft', serverId: 'srv_hb', name: '草稿', protocol: 'vless-reality-vision', topology: 'direct', networkMode: 'ipv4', port: 23457, resourceId: 'res_draft', status: 'draft', realityServerName: 'www.tesla.com', createdAt: stamp, updatedAt: stamp }, + { id: 'nod_idle', serverId: 'srv_hb', name: '无客户', protocol: 'vless-reality-vision', topology: 'direct', networkMode: 'ipv4', port: 23458, resourceId: 'res_idle', status: 'active', entryReady: true, realityServerName: 'www.tesla.com', createdAt: stamp, updatedAt: stamp }, + { id: 'nod_rm', serverId: 'srv_hb', name: '删除中', protocol: 'vless-reality-vision', topology: 'direct', networkMode: 'ipv4', port: 23459, resourceId: 'res_rm', status: 'removing', entryReady: true, realityServerName: 'www.tesla.com', createdAt: stamp, updatedAt: stamp }, + ); + data.nodeClients.push( + { id: 'ncl_draft', nodeId: 'nod_draft', customerId: 'cus_hb', status: 'active', credentials: { clientId: crypto.randomUUID(), relayPassword: 'x', shortId: 'aa' }, createdAt: stamp, updatedAt: stamp }, + { id: 'ncl_rm', nodeId: 'nod_rm', customerId: 'cus_hb', status: 'active', credentials: { clientId: crypto.randomUUID(), relayPassword: 'x', shortId: 'bb' }, createdAt: stamp, updatedAt: stamp }, + ); + }); + // Only the in-scope node (nod_hb) is requeued; the heartbeat holds nothing. + assert.equal(await orchestrator.reconcileFromHeartbeat('srv_hb', []), 1); + assert.deepEqual(applyJobs(store).map((job) => job.nodeId), ['nod_hb']); +}); + +test('heartbeat reconcile requeues a forward node whose exit resource is missing', async (t) => { + const { store, orchestrator } = await fixture(t); + const stamp = new Date().toISOString(); + await store.transaction((data) => { + data.servers.push({ id: 'srv_exit', name: '出口', publicAddress: '203.0.113.20', portRangeStart: 20000, portRangeEnd: 50000, status: 'online', lastSeenAt: stamp }); + data.nodes.push({ + id: 'nod_fwd', serverId: 'srv_hb', name: '中转节点', protocol: 'vless-reality-vision', topology: 'forward', + exitServerId: 'srv_exit', exitProtocol: 'shadowsocks-2022-aes128', exitPort: 23460, exitResourceId: 'res_fwd_exit', + exitCredentials: { password: 'exit-secret' }, + networkMode: 'ipv4', port: 23461, resourceId: 'res_fwd_in', + status: 'active', entryReady: true, exitReady: true, realityServerName: 'www.tesla.com', createdAt: stamp, updatedAt: stamp, + }); + data.nodeClients.push({ + id: 'ncl_fwd', nodeId: 'nod_fwd', customerId: 'cus_hb', status: 'active', + credentials: { clientId: crypto.randomUUID(), relayPassword: 'secret', shortId: 'ccdd1234' }, + createdAt: stamp, updatedAt: stamp, + }); + }); + // The exit server's agent holds nothing: the exit resource is re-dispatched. + assert.equal(await orchestrator.reconcileFromHeartbeat('srv_exit', []), 1); + const exitJobs = applyJobs(store, 'nod_fwd').filter((job) => job.serverId === 'srv_exit'); + assert.equal(exitJobs.length, 1); + assert.equal(exitJobs[0].payload.kind, 'node-exit'); + // The entry server's agent holds its resource: nothing is queued for it. + assert.equal(applyJobs(store, 'nod_fwd').filter((job) => job.serverId === 'srv_hb').length, 0); +}); + +async function freePort() { + return new Promise((resolve, reject) => { + const server = net.createServer(); + server.once('error', reject); + server.listen(0, '127.0.0.1', () => { const port = server.address().port; server.close(() => resolve(port)); }); + }); +} + +async function waitFor(url) { + for (let i = 0; i < 60; i += 1) { + try { const response = await fetch(url); if (response.ok) return; } catch { /* retry */ } + await new Promise((resolve) => setTimeout(resolve, 100)); + } + throw new Error('test server did not start'); +} + +test('E2E: wiped agent state is healed by a heartbeat reporting no resources', async (t) => { + const dir = await fs.promises.mkdtemp(path.join(os.tmpdir(), 'nexusgate-hb-e2e-')); + const port = await freePort(); + const child = spawn(process.execPath, ['server.js'], { + cwd: path.resolve(__dirname, '..'), + env: { ...process.env, NG_HOST: '127.0.0.1', NG_PORT: String(port), NG_DATA_FILE: path.join(dir, 'data.json'), NG_ADMIN_PASSWORD: 'test-password', NG_COOKIE_SECURE: 'false' }, + stdio: ['ignore', 'pipe', 'pipe'], + }); + t.after(async () => { child.kill('SIGTERM'); await fs.promises.rm(dir, { recursive: true, force: true }); }); + const base = `http://127.0.0.1:${port}`; + await waitFor(`${base}/healthz`); + + const login = await fetch(`${base}/api/auth/login`, { method: 'POST', headers: { 'content-type': 'application/json' }, body: JSON.stringify({ username: 'admin', password: 'test-password' }) }); + assert.equal(login.status, 200); + const session = await login.json(); + const cookie = login.headers.get('set-cookie').split(';')[0]; + const request = async (pathname, method = 'GET', body) => { + const response = await fetch(`${base}${pathname}`, { method, headers: { cookie, 'x-csrf-token': session.csrf, ...(body ? { 'content-type': 'application/json' } : {}) }, body: body ? JSON.stringify(body) : undefined }); + const json = await response.json(); + assert.ok(response.ok, JSON.stringify(json)); + return json; + }; + + const server = (await request('/api/servers', 'POST', { name: '入口 01', role: 'relay', publicAddress: 'entry.example.com' })).server; + const customer = (await request('/api/customers', 'POST', { name: '客户 A' })).customer; + const ticket = await request(`/api/servers/${server.id}/enrollment-token`, 'POST', {}); + const enrolled = await fetch(`${base}/api/agent/enroll`, { method: 'POST', headers: { 'content-type': 'application/json' }, body: JSON.stringify({ token: ticket.token, hostname: 'sim-agent', version: '0.6.8' }) }); + assert.equal(enrolled.status, 201); + const agentKey = (await enrolled.json()).agentKey; + const agentHeaders = { authorization: `Bearer ${agentKey}`, 'content-type': 'application/json' }; + const heartbeat = async (body) => { + const response = await fetch(`${base}/api/agent/heartbeat`, { method: 'POST', headers: agentHeaders, body: JSON.stringify(body) }); + assert.equal(response.status, 200); + }; + const poll = async () => { + const response = await fetch(`${base}/api/agent/poll`, { method: 'POST', headers: agentHeaders, body: '{}' }); + assert.equal(response.status, 200); + return (await response.json()).job; + }; + + await heartbeat({ version: '0.6.8', system: {} }); + const node = (await request('/api/nodes', 'POST', { name: 'Reality 直连', serverId: server.id, protocol: 'vless-reality-vision', topology: 'direct', realityServerName: 'www.amd.com' })).node; + await request(`/api/nodes/${node.id}/assign`, 'POST', { customerId: customer.id }); + await request(`/api/nodes/${node.id}/deploy`, 'POST', {}); + + const keyId = `reality-${node.id}`; + const first = await poll(); + assert.ok(first, 'initial deploy job was dispatched'); + assert.equal(first.action, 'apply_resource'); + assert.match(first.payload.keyMaterial[keyId].privateKey, B64URL_43); + const completed = await fetch(`${base}/api/agent/jobs/${first.id}/complete`, { method: 'POST', headers: agentHeaders, body: JSON.stringify({ success: true, result: { artifacts: {} } }) }); + assert.equal(completed.status, 200); + assert.equal(await poll(), null); + + // The node's local state is wiped and only the agent restarts: its next + // heartbeat reports no resources, and the panel must re-dispatch on its own. + await heartbeat({ version: '0.6.8', system: {}, resources: [] }); + const healed = await poll(); + assert.ok(healed, 'heartbeat with empty resources triggered a re-dispatch'); + assert.equal(healed.action, 'apply_resource'); + assert.equal(healed.nodeId, node.id); + assert.equal(healed.payload.keyMaterial[keyId].privateKey, first.payload.keyMaterial[keyId].privateKey); + assert.equal(healed.payload.keyMaterial[keyId].publicKey, first.payload.keyMaterial[keyId].publicKey); +});