feat: heartbeat reconciliation — panel re-dispatches resources a node lost
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.
This commit is contained in:
aaaamoooo
+86
-32
@@ -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 };
|
||||
|
||||
Reference in New Issue
Block a user