858 lines
37 KiB
JavaScript
858 lines
37 KiB
JavaScript
import assert from 'node:assert/strict';
|
|
import {readFile} from 'node:fs/promises';
|
|
import {setImmediate} from 'node:timers/promises';
|
|
import vm from 'node:vm';
|
|
import test from 'node:test';
|
|
import {transformSync} from 'esbuild';
|
|
import {Buffer} from 'node:buffer';
|
|
|
|
const source = await readFile(new URL('../index.ts', import.meta.url), 'utf8');
|
|
const code = transformSync(source.replace(/^import .*;\n/gm, ''), {loader: 'ts'}).code;
|
|
const configured = JSON.parse(await readFile(new URL('../config.json', import.meta.url), 'utf8'));
|
|
const relay_test_key = 'node-registration-test-secret-not-production';
|
|
const success = node_session_id => ({status: 'success', data: {name: 'Node', node_session_id}});
|
|
const deferred = () => {
|
|
let resolve, reject;
|
|
const promise = new Promise((ok, fail) => { resolve = ok; reject = fail; });
|
|
return {promise, resolve, reject};
|
|
};
|
|
|
|
async function setup({forward = false, start, plugins, functions, workflow_catalog, versions, status, init, connect_socket, stop, contexts = [], stop_bot,
|
|
node = configured.node, auto_reconnect = {...configured.main_server.auto_reconnect, interval: 100}} = {}) {
|
|
const handlers = new Map();
|
|
const signals = new Map();
|
|
const timers = new Map();
|
|
const timer_delays = new Map();
|
|
const deadlines = new Map();
|
|
const exits = [];
|
|
const stopped_bots = [];
|
|
const requests = [];
|
|
const bindings = [];
|
|
const logs = [];
|
|
const updates = [];
|
|
const workflow_reports = [];
|
|
const controls = [];
|
|
const relay_options = [];
|
|
const order = [];
|
|
let timer_id = 0;
|
|
let any;
|
|
let shutting_down = false;
|
|
const state = {sockets: 0, connects: 0, disconnects: 0, clears: 0, stops: 0, releases: 0, events: 0};
|
|
const socket = {
|
|
connected: false,
|
|
on(event, handler) {
|
|
if (!handlers.has(event)) handlers.set(event, new Set());
|
|
handlers.get(event).add(handler);
|
|
},
|
|
off(event, handler) { handlers.get(event)?.delete(handler); },
|
|
onAny(handler) { any = handler; },
|
|
connect() { state.connects++; connect_socket?.(); },
|
|
disconnect() {
|
|
state.disconnects++;
|
|
if (!socket.connected) return;
|
|
socket.connected = false;
|
|
dispatch('disconnect', 'io client disconnect');
|
|
},
|
|
timeout(ms) { assert.equal(ms, 5000); return socket; },
|
|
emitWithAck(event, payload) {
|
|
const pending = deferred();
|
|
requests.push({event, payload, ...pending});
|
|
order.push(event);
|
|
return pending.promise;
|
|
},
|
|
emit() {},
|
|
};
|
|
function dispatch(event, ...args) {
|
|
for (const handler of [...(handlers.get(event) ?? [])]) handler(...args);
|
|
}
|
|
let release;
|
|
const handle = {
|
|
bind_control(control) {
|
|
release?.();
|
|
controls.push(control);
|
|
order.push('bind_control');
|
|
const listener = (...args) => { state.events++; args.at(-1)?.(); };
|
|
control.on('forward_open_session', listener);
|
|
let active = true;
|
|
release = () => {
|
|
if (!active) return;
|
|
active = false;
|
|
state.releases++;
|
|
control.off('forward_open_session', listener);
|
|
};
|
|
return release;
|
|
},
|
|
async stop() { state.stops++; release?.(); await stop?.(); order.push('relay_stopped'); },
|
|
};
|
|
const process = {
|
|
once: (name, handler) => signals.set(name, handler),
|
|
exit: code => { exits.push(code); order.push('exit'); },
|
|
};
|
|
const config = {
|
|
...configured, node: {...node},
|
|
main_server: {...configured.main_server, auto_reconnect},
|
|
forward_server: {...configured.forward_server, key: relay_test_key, status: forward},
|
|
};
|
|
vm.runInNewContext(code, {
|
|
Buffer,
|
|
config,
|
|
io: (_url, options) => {
|
|
assert.equal(options.reconnection, false);
|
|
state.sockets++;
|
|
return socket;
|
|
},
|
|
init_event_handler_list: async () => { await init?.(); },
|
|
get_bot_context_status: async () => contexts,
|
|
stop_bot_context: async bot_id => {
|
|
stopped_bots.push(bot_id);
|
|
await stop_bot?.(bot_id);
|
|
order.push(`bot_stopped:${bot_id}`);
|
|
},
|
|
bind_node_session: (bound_socket, node_id, session) => {
|
|
assert.equal(bound_socket, socket);
|
|
bindings.push({node_id, session});
|
|
},
|
|
clear_node_session: bound_socket => {
|
|
assert.equal(bound_socket, socket);
|
|
state.clears++;
|
|
},
|
|
update_plugins: async () => { updates.push('plugins'); order.push('plugins'); await plugins?.(); },
|
|
update_function_list: async () => { updates.push('functions'); order.push('functions'); await functions?.(); },
|
|
update_workflow_catalog: async (bound_socket, node_id, is_current) => {
|
|
assert.equal(bound_socket, socket);
|
|
assert.equal(node_id, config.node.id);
|
|
assert.equal(is_current(), true);
|
|
workflow_reports.push({is_current});
|
|
updates.push('workflow_catalog');
|
|
order.push('workflow_catalog');
|
|
await workflow_catalog?.();
|
|
},
|
|
update_version_list: async () => { updates.push('versions'); order.push('versions'); await versions?.(); },
|
|
update_all_bot_status: async () => { updates.push('bots'); order.push('bots'); await status?.(); },
|
|
event_handler: async () => { state.events++; },
|
|
start_forward_server: async options => {
|
|
relay_options.push(options);
|
|
order.push('start');
|
|
const ready = start ? await start(handle) : handle;
|
|
ready.bind_control(options.control);
|
|
return ready;
|
|
},
|
|
console_print: {
|
|
log: message => logs.push(message),
|
|
info: message => message, error: message => message,
|
|
success: message => message, warning: message => message,
|
|
},
|
|
process,
|
|
setTimeout: (callback, delay) => {
|
|
if (shutting_down) {
|
|
assert.equal(delay, 10_000);
|
|
deadlines.set(++timer_id, callback);
|
|
return timer_id;
|
|
}
|
|
const interval = auto_reconnect?.interval;
|
|
assert.equal(delay, Number.isInteger(interval) && interval > 0 && interval <= 2_147_483_647 ? interval : 10_000);
|
|
timers.set(++timer_id, callback);
|
|
timer_delays.set(timer_id, delay);
|
|
return timer_id;
|
|
},
|
|
clearTimeout: id => { timers.delete(id); timer_delays.delete(id); deadlines.delete(id); },
|
|
});
|
|
await setImmediate();
|
|
return {
|
|
state, socket, process, config, timers, timer_delays, requests, bindings, logs, updates, workflow_reports, controls,
|
|
relay_options, order, handle, dispatch, deadlines, exits, stopped_bots,
|
|
connect() { socket.connected = true; dispatch('connect'); },
|
|
disconnect() { socket.connected = false; dispatch('disconnect', 'transport close'); },
|
|
event(event, ...args) { any(event, ...args); },
|
|
signal(name) { shutting_down = true; signals.get(name)(); },
|
|
tick() {
|
|
assert.equal(timers.size, 1);
|
|
const [id, callback] = timers.entries().next().value;
|
|
timers.delete(id);
|
|
timer_delays.delete(id);
|
|
callback();
|
|
},
|
|
};
|
|
}
|
|
|
|
test('registration binds the exact structured ACK and replaces sessions on reconnect and re-authentication', async () => {
|
|
const app = await setup();
|
|
app.event('bot_start', {});
|
|
assert.equal(app.state.events, 0);
|
|
app.connect();
|
|
assert.equal(app.state.clears, 1);
|
|
assert.equal(app.requests[0].event, 'register_node');
|
|
app.requests[0].resolve(success('first-session'));
|
|
await setImmediate();
|
|
assert.deepEqual(app.bindings, [{node_id: configured.node.id, session: 'first-session'}]);
|
|
const stages = ['plugins', 'functions', 'workflow_catalog', 'versions', 'bots'];
|
|
assert.deepEqual(app.updates, stages);
|
|
assert.deepEqual(app.order, ['register_node', ...stages]);
|
|
assert.equal(app.workflow_reports.length, 1);
|
|
assert.equal(app.workflow_reports[0].is_current(), true);
|
|
app.event('bot_start', {});
|
|
assert.equal(app.state.events, 1);
|
|
app.disconnect();
|
|
assert.equal(app.state.clears, 2);
|
|
assert.equal(app.workflow_reports[0].is_current(), false);
|
|
app.event('bot_start', {});
|
|
assert.equal(app.state.events, 1);
|
|
app.tick();
|
|
app.connect();
|
|
app.requests[1].resolve(success('second-session'));
|
|
await setImmediate();
|
|
assert.deepEqual(app.updates, [...stages, ...stages]);
|
|
assert.equal(app.workflow_reports.length, 2);
|
|
assert.equal(app.workflow_reports[1].is_current(), true);
|
|
app.connect();
|
|
assert.equal(app.state.clears, 4);
|
|
app.requests[2].resolve(success('third-session'));
|
|
await setImmediate();
|
|
assert.deepEqual(app.bindings.map(binding => binding.session), ['first-session', 'second-session', 'third-session']);
|
|
assert.deepEqual(app.updates, [...stages, ...stages, ...stages]);
|
|
assert.equal(app.workflow_reports.length, 3);
|
|
assert.equal(app.workflow_reports[1].is_current(), false);
|
|
assert.equal(app.workflow_reports[2].is_current(), true);
|
|
});
|
|
|
|
test('workflow catalog reporting is awaited before versions, Bot status, and relay startup', async () => {
|
|
const pending = deferred();
|
|
const app = await setup({forward: true, workflow_catalog: () => pending.promise});
|
|
app.connect();
|
|
app.requests[0].resolve(success('session'));
|
|
await setImmediate();
|
|
assert.deepEqual(app.updates, ['plugins', 'functions', 'workflow_catalog']);
|
|
assert.equal(app.workflow_reports.length, 1);
|
|
assert.equal(app.relay_options.length, 0);
|
|
assert.equal(app.requests.length, 1);
|
|
pending.resolve();
|
|
await setImmediate();
|
|
assert.deepEqual(app.updates, ['plugins', 'functions', 'workflow_catalog', 'versions', 'bots']);
|
|
assert.deepEqual(app.order, ['register_node', ...app.updates, 'start', 'bind_control', 'bind_control', 'register_forward']);
|
|
assert.equal(app.workflow_reports.length, 1);
|
|
app.requests[1].resolve({status: 'success', data: 'Relay'});
|
|
await setImmediate();
|
|
assert.equal(app.timers.size, 0);
|
|
});
|
|
|
|
test('workflow catalog rejection and ACK timeout retry registration and report again after recovery', async () => {
|
|
for (const failure of ['report rejected secret-token', 'ACK timeout secret-token']) {
|
|
const pending = deferred();
|
|
let calls = 0;
|
|
const app = await setup({forward: true, contexts: [{bot_id: 1}],
|
|
workflow_catalog: () => ++calls === 1 ? pending.promise : undefined});
|
|
app.connect();
|
|
app.requests[0].resolve(success('failed-session'));
|
|
await setImmediate();
|
|
pending.reject(new Error(failure));
|
|
await setImmediate();
|
|
assert.deepEqual(app.updates, ['plugins', 'functions', 'workflow_catalog']);
|
|
assert.equal(app.relay_options.length, 0);
|
|
assert.equal(app.socket.connected, false);
|
|
assert.equal(app.state.disconnects, 1);
|
|
assert.equal(app.timers.size, 1);
|
|
assert.equal(app.workflow_reports[0].is_current(), false);
|
|
assert.ok(app.logs.some(message => message.includes('阶段: 工作流目录上报')));
|
|
assert.ok(!app.logs.join('\n').includes('secret-token'));
|
|
assert.equal(app.state.stops, 0);
|
|
assert.deepEqual(app.stopped_bots, []);
|
|
assert.deepEqual(app.exits, []);
|
|
assert.equal(app.process.exitCode, undefined);
|
|
app.tick();
|
|
app.connect();
|
|
assert.equal(app.requests[1].event, 'register_node');
|
|
app.requests[1].resolve(success('recovered-session'));
|
|
await setImmediate();
|
|
assert.equal(calls, 2);
|
|
assert.equal(app.workflow_reports.length, 2);
|
|
assert.deepEqual(app.updates, ['plugins', 'functions', 'workflow_catalog',
|
|
'plugins', 'functions', 'workflow_catalog', 'versions', 'bots']);
|
|
assert.equal(app.requests[2].event, 'register_forward');
|
|
app.requests[2].resolve({status: 'success', data: 'Relay'});
|
|
await setImmediate();
|
|
assert.equal(app.socket.connected, true);
|
|
assert.equal(app.timers.size, 0);
|
|
assert.equal(app.state.stops, 0);
|
|
assert.deepEqual(app.stopped_bots, []);
|
|
assert.deepEqual(app.exits, []);
|
|
}
|
|
});
|
|
|
|
test('stale workflow catalog completion or failure cannot continue stages or disconnect a newer session', async () => {
|
|
for (const rejected of [false, true]) {
|
|
const pending = deferred();
|
|
let calls = 0;
|
|
const app = await setup({forward: true,
|
|
workflow_catalog: () => ++calls === 1 ? pending.promise : undefined});
|
|
app.connect();
|
|
app.requests[0].resolve(success('old-session'));
|
|
await setImmediate();
|
|
assert.deepEqual(app.updates, ['plugins', 'functions', 'workflow_catalog']);
|
|
app.disconnect();
|
|
app.tick();
|
|
app.connect();
|
|
app.requests[1].resolve(success('new-session'));
|
|
await setImmediate();
|
|
app.requests[2].resolve({status: 'success', data: 'Relay'});
|
|
await setImmediate();
|
|
assert.equal(app.workflow_reports[0].is_current(), false);
|
|
assert.equal(app.workflow_reports[1].is_current(), true);
|
|
const order = [...app.order];
|
|
if (rejected) pending.reject(new Error('ACK timeout secret-token'));
|
|
else pending.resolve();
|
|
await setImmediate();
|
|
assert.deepEqual(app.order, order);
|
|
assert.deepEqual(app.updates, ['plugins', 'functions', 'workflow_catalog',
|
|
'plugins', 'functions', 'workflow_catalog', 'versions', 'bots']);
|
|
assert.equal(app.workflow_reports.length, 2);
|
|
assert.equal(app.socket.connected, true);
|
|
assert.equal(app.state.disconnects, 0);
|
|
assert.equal(app.timers.size, 0);
|
|
assert.deepEqual(app.bindings.map(binding => binding.session), ['old-session', 'new-session']);
|
|
assert.equal(app.relay_options.length, 1);
|
|
assert.equal(app.requests.length, 3);
|
|
assert.equal(app.process.exitCode, undefined);
|
|
assert.ok(!app.logs.join('\n').includes('阶段: 工作流目录上报'));
|
|
assert.ok(!app.logs.join('\n').includes('secret-token'));
|
|
}
|
|
});
|
|
|
|
test('failed relay startup reports a safe reason, retries startup, and registers after recovery', async () => {
|
|
let attempts = 0;
|
|
const app = await setup({forward: true, contexts: [{bot_id: 1}], start: async handle => {
|
|
if (++attempts === 1) throw new Error('A relay secret of at least 16 characters is required');
|
|
return handle;
|
|
}});
|
|
app.connect();
|
|
app.requests[0].resolve(success('session'));
|
|
await setImmediate();
|
|
assert.ok(app.logs.some(message => message.includes('阶段: 转发服务器启动') && message.includes('至少需要 16 个字符')));
|
|
assert.ok(!app.logs.some(message => message.includes('节点关闭失败或超时')));
|
|
assert.equal(app.state.stops, 0);
|
|
assert.equal(app.deadlines.size, 0);
|
|
assert.equal(app.process.exitCode, undefined);
|
|
assert.equal(app.timers.size, 1);
|
|
assert.deepEqual(app.stopped_bots, []);
|
|
app.tick();
|
|
app.connect();
|
|
app.requests[1].resolve(success('recovered-session'));
|
|
await setImmediate();
|
|
assert.equal(attempts, 2);
|
|
assert.equal(app.relay_options.length, 2);
|
|
assert.equal(app.requests[2].event, 'register_forward');
|
|
app.requests[2].resolve({status: 'success', data: 'Relay'});
|
|
await setImmediate();
|
|
assert.equal(app.timers.size, 0);
|
|
assert.equal(app.state.stops, 0);
|
|
assert.deepEqual(app.stopped_bots, []);
|
|
assert.deepEqual(app.exits, []);
|
|
});
|
|
|
|
test('relay listen errors disclose only safe error category and initialization stage', async () => {
|
|
for (const code of ['EADDRINUSE', 'EACCES']) {
|
|
const app = await setup({forward: true, start: async () => {
|
|
throw Object.assign(new Error('SECRET config/token/stack'), {code});
|
|
}});
|
|
app.connect();
|
|
app.requests[0].resolve(success('session'));
|
|
await setImmediate();
|
|
assert.ok(app.logs.some(message => message.includes('阶段: 转发服务器启动')
|
|
&& message.includes(code === 'EADDRINUSE' ? '端口已被占用' : '没有权限')));
|
|
assert.ok(!app.logs.join('\n').includes('SECRET'));
|
|
assert.ok(!app.logs.join('\n').includes('节点关闭失败或超时'));
|
|
assert.equal(app.timers.size, 1);
|
|
assert.equal(app.process.exitCode, undefined);
|
|
}
|
|
});
|
|
|
|
test('legacy, missing and extra registration fields never authenticate and retry safely', async () => {
|
|
for (const response of [
|
|
{status: 'success', data: 'legacy name'},
|
|
{status: 'success', data: {name: 'Node'}},
|
|
{status: 'success', data: {name: 42, node_session_id: 'session'}},
|
|
{status: 'success', data: {name: 'Node', node_session_id: ' '}},
|
|
success('x'.repeat(257)),
|
|
success('\u00e9'.repeat(129)),
|
|
{status: 'success', data: {name: 'Node', node_session_id: 'session', token: 'secret-token'}},
|
|
{...success('session'), token: 'secret-token'},
|
|
]) {
|
|
const app = await setup();
|
|
app.connect();
|
|
app.requests[0].resolve(response);
|
|
await setImmediate();
|
|
assert.equal(app.bindings.length, 0);
|
|
assert.equal(app.updates.length, 0);
|
|
assert.equal(app.timers.size, 1);
|
|
assert.equal(app.process.exitCode, undefined);
|
|
assert.ok(!app.logs.join('\n').includes('secret-token'));
|
|
}
|
|
});
|
|
|
|
test('registration accepts a 256-byte UTF-8 session ID and rejects unsafe configured Node IDs', async () => {
|
|
const app = await setup();
|
|
app.connect();
|
|
app.requests[0].resolve(success('\u00e9'.repeat(128)));
|
|
await setImmediate();
|
|
assert.equal(app.bindings[0].session, '\u00e9'.repeat(128));
|
|
const invalid = await setup({node: {...configured.node, id: Number.MAX_SAFE_INTEGER + 1}});
|
|
assert.equal(invalid.state.connects, 0);
|
|
assert.equal(invalid.state.sockets, 0);
|
|
assert.equal(invalid.requests.length, 0);
|
|
assert.equal(invalid.timers.size, 1);
|
|
assert.equal(invalid.process.exitCode, undefined);
|
|
invalid.config.node.id = configured.node.id;
|
|
invalid.tick();
|
|
await setImmediate();
|
|
assert.equal(invalid.state.sockets, 1);
|
|
assert.equal(invalid.state.connects, 1);
|
|
});
|
|
|
|
test('handler loading failures retry bootstrap without creating a socket and recover after success', async () => {
|
|
let attempts = 0;
|
|
const app = await setup({init: async () => {
|
|
if (++attempts <= 3) throw new Error('secret-token');
|
|
}});
|
|
for (let attempt = 1; attempt <= 3; attempt++) {
|
|
assert.equal(attempts, attempt);
|
|
assert.equal(app.state.sockets, 0);
|
|
assert.equal(app.state.connects, 0);
|
|
assert.equal(app.requests.length, 0);
|
|
assert.equal(app.timers.size, 1);
|
|
assert.equal(app.process.exitCode, undefined);
|
|
app.tick();
|
|
await setImmediate();
|
|
}
|
|
assert.equal(attempts, 4);
|
|
assert.equal(app.state.sockets, 1);
|
|
assert.equal(app.state.connects, 1);
|
|
assert.equal(app.timers.size, 0);
|
|
assert.deepEqual(app.exits, []);
|
|
assert.ok(!app.logs.join('\n').includes('secret-token'));
|
|
});
|
|
|
|
test('missing or invalid reconnect settings use the default interval and keep retrying', async () => {
|
|
for (const auto_reconnect of [null, {}, {interval: 0}, {interval: -1}, {interval: 1.5},
|
|
{interval: '100'}, {interval: 2_147_483_648}, {enabled: false, max_attempts: -1, interval: 250}]) {
|
|
const app = await setup({auto_reconnect});
|
|
assert.equal(app.state.sockets, 1);
|
|
assert.equal(app.state.connects, 1);
|
|
app.dispatch('connect_error', new Error('secret-token'));
|
|
assert.deepEqual([...app.timer_delays.values()], [auto_reconnect?.interval === 250 ? 250 : 10_000]);
|
|
app.tick();
|
|
app.dispatch('connect_error', new Error('secret-token'));
|
|
assert.equal(app.state.connects, 2);
|
|
assert.equal(app.timers.size, 1);
|
|
assert.equal(app.process.exitCode, undefined);
|
|
assert.deepEqual(app.exits, []);
|
|
}
|
|
});
|
|
|
|
test('synchronous initial and recurring connect failures keep retrying until connection succeeds', async () => {
|
|
let attempts = 0;
|
|
const app = await setup({auto_reconnect: {enabled: false, max_attempts: 0, interval: 100},
|
|
connect_socket: () => {
|
|
if (++attempts <= 4) throw new Error('secret-token');
|
|
}});
|
|
for (let attempt = 1; attempt <= 4; attempt++) {
|
|
assert.equal(app.state.connects, attempt);
|
|
assert.equal(app.timers.size, 1);
|
|
assert.equal(app.process.exitCode, undefined);
|
|
app.tick();
|
|
}
|
|
assert.equal(app.state.connects, 5);
|
|
assert.equal(app.state.sockets, 1);
|
|
assert.equal(app.timers.size, 0);
|
|
app.connect();
|
|
app.requests[0].resolve(success('recovered-session'));
|
|
await setImmediate();
|
|
assert.deepEqual(app.bindings.map(binding => binding.session), ['recovered-session']);
|
|
assert.deepEqual(app.exits, []);
|
|
assert.ok(!app.logs.join('\n').includes('secret-token'));
|
|
});
|
|
|
|
test('plugin, function, version, and Bot status failures retry initialization and recover without stopping Bots', async () => {
|
|
for (const stage of ['plugins', 'functions', 'versions', 'status']) {
|
|
let calls = 0;
|
|
const app = await setup({contexts: [{bot_id: 1}], [stage]: async () => {
|
|
if (++calls === 1) throw new Error('secret-token');
|
|
}});
|
|
app.connect();
|
|
app.requests[0].resolve(success('failed-session'));
|
|
await setImmediate();
|
|
assert.equal(app.socket.connected, false);
|
|
assert.equal(app.timers.size, 1);
|
|
assert.equal(app.state.stops, 0);
|
|
assert.deepEqual(app.stopped_bots, []);
|
|
assert.equal(app.process.exitCode, undefined);
|
|
assert.ok(!app.logs.join('\n').includes('secret-token'));
|
|
app.event('bot_start', {});
|
|
assert.equal(app.state.events, 0);
|
|
app.tick();
|
|
app.connect();
|
|
app.requests[1].resolve(success('recovered-session'));
|
|
await setImmediate();
|
|
assert.equal(calls, 2);
|
|
assert.deepEqual(app.updates.slice(-5), ['plugins', 'functions', 'workflow_catalog', 'versions', 'bots']);
|
|
assert.equal(app.timers.size, 0);
|
|
assert.equal(app.socket.connected, true);
|
|
app.event('bot_start', {});
|
|
assert.equal(app.state.events, 1);
|
|
assert.deepEqual(app.stopped_bots, []);
|
|
assert.deepEqual(app.exits, []);
|
|
}
|
|
});
|
|
|
|
test('node registration refusals and ACK timeouts continue retrying beyond disabled and exhausted settings', async () => {
|
|
const app = await setup({contexts: [{bot_id: 1}],
|
|
auto_reconnect: {enabled: false, max_attempts: 1, interval: 100}});
|
|
for (let attempt = 0; attempt < 6; attempt++) {
|
|
app.connect();
|
|
const request = app.requests[attempt];
|
|
assert.equal(request.event, 'register_node');
|
|
if (attempt % 2) request.reject(new Error('ACK timeout secret-token'));
|
|
else request.resolve({status: 'error', data: 401});
|
|
await setImmediate();
|
|
assert.equal(app.bindings.length, 0);
|
|
assert.equal(app.updates.length, 0);
|
|
assert.equal(app.timers.size, 1);
|
|
assert.equal(app.process.exitCode, undefined);
|
|
assert.deepEqual(app.stopped_bots, []);
|
|
app.tick();
|
|
}
|
|
app.connect();
|
|
app.requests[6].resolve(success('recovered-session'));
|
|
await setImmediate();
|
|
assert.deepEqual(app.bindings.map(binding => binding.session), ['recovered-session']);
|
|
assert.equal(app.timers.size, 0);
|
|
assert.deepEqual(app.exits, []);
|
|
assert.ok(!app.logs.join('\n').includes('secret-token'));
|
|
});
|
|
|
|
test('stale registration ACKs and failures cannot replace a newer session or terminate it', async () => {
|
|
for (const rejected of [false, true]) {
|
|
const app = await setup();
|
|
app.connect();
|
|
app.disconnect();
|
|
app.tick();
|
|
app.connect();
|
|
app.requests[1].resolve(success('current-session'));
|
|
await setImmediate();
|
|
if (rejected) app.requests[0].reject(new Error('secret-token'));
|
|
else app.requests[0].resolve(success('obsolete-session'));
|
|
await setImmediate();
|
|
assert.deepEqual(app.bindings.map(binding => binding.session), ['current-session']);
|
|
assert.equal(app.state.disconnects, 0);
|
|
assert.equal(app.process.exitCode, undefined);
|
|
assert.ok(!app.logs.join('\n').includes('secret-token'));
|
|
}
|
|
});
|
|
|
|
test('registration refusal and transport errors do not log response objects, URLs, or tokens', async () => {
|
|
const app = await setup();
|
|
const secret = 'https://private.example/?token=secret-token';
|
|
app.dispatch('connect_error', new Error(secret));
|
|
app.dispatch('error', {error: secret});
|
|
app.connect();
|
|
app.requests[0].resolve({status: 'error', data: {token: secret}});
|
|
await setImmediate();
|
|
assert.equal(app.process.exitCode, undefined);
|
|
assert.equal(app.timers.size, 1);
|
|
assert.equal(app.bindings.length, 0);
|
|
assert.ok(!app.logs.join('\n').includes(secret));
|
|
assert.ok(!app.logs.join('\n').includes(configured.main_server.host));
|
|
});
|
|
|
|
test('hosted relay starts once, registers only when ready, rebinds control, and accepts other target Nodes', async () => {
|
|
const ready = deferred();
|
|
const app = await setup({forward: true, start: () => ready.promise});
|
|
app.connect();
|
|
app.requests[0].resolve(success('host-session'));
|
|
await setImmediate();
|
|
assert.equal(app.requests.length, 1);
|
|
assert.equal(app.relay_options.length, 1);
|
|
const options = app.relay_options[0];
|
|
assert.equal(options.forward_id, configured.forward_server.id);
|
|
assert.equal(options.key, relay_test_key);
|
|
assert.ok(options.key.length >= 32);
|
|
assert.deepEqual(options.allowed_origins, configured.forward_server.allowed_origins);
|
|
assert.equal(options.path, configured.forward_server.path);
|
|
assert.equal(options.namespace, configured.forward_server.namespace);
|
|
assert.equal(options.http_server, undefined);
|
|
ready.resolve(app.handle);
|
|
await setImmediate();
|
|
assert.equal(app.requests[1].event, 'register_forward');
|
|
assert.ok(app.order.indexOf('bind_control') < app.order.indexOf('register_forward'));
|
|
const stale_control_callback = app.controls.at(-1);
|
|
const listener = () => app.state.events++;
|
|
stale_control_callback.on('test_control', listener);
|
|
app.dispatch('forward_open_session', {node_id: 900, node_session_id: 'other-node-session'}, () => {});
|
|
assert.equal(app.state.events, 1);
|
|
app.requests[1].resolve({status: 'success', data: 'Relay'});
|
|
await setImmediate();
|
|
app.disconnect();
|
|
app.dispatch('test_control');
|
|
assert.equal(app.state.events, 1);
|
|
app.tick();
|
|
app.connect();
|
|
app.requests[2].resolve(success('host-session-2'));
|
|
await setImmediate();
|
|
app.dispatch('test_control');
|
|
assert.equal(app.state.events, 1);
|
|
assert.equal(app.relay_options.length, 1);
|
|
assert.equal(app.requests[3].event, 'register_forward');
|
|
app.dispatch('forward_open_session', {node_id: 901, node_session_id: 'third-node-session'}, () => {});
|
|
assert.equal(app.state.events, 2);
|
|
stale_control_callback.off('test_control', listener);
|
|
app.signal('SIGINT');
|
|
await setImmediate();
|
|
assert.equal(app.state.stops, 1);
|
|
assert.equal(app.process.exitCode, 0);
|
|
});
|
|
|
|
test('disconnect during relay startup does not duplicate startup or register the stale connection', async () => {
|
|
const ready = deferred();
|
|
const app = await setup({forward: true, start: () => ready.promise});
|
|
app.connect();
|
|
app.requests[0].resolve(success('old-session'));
|
|
await setImmediate();
|
|
app.disconnect();
|
|
app.tick();
|
|
app.connect();
|
|
app.requests[1].resolve(success('new-session'));
|
|
await setImmediate();
|
|
ready.resolve(app.handle);
|
|
await setImmediate();
|
|
assert.equal(app.relay_options.length, 1);
|
|
assert.equal(app.requests.filter(request => request.event === 'register_forward').length, 1);
|
|
assert.equal(app.state.stops, 0);
|
|
});
|
|
|
|
test('SIGTERM waits for pending relay startup, stops its handle, and never registers', async () => {
|
|
const ready = deferred();
|
|
const app = await setup({forward: true, start: () => ready.promise});
|
|
app.connect();
|
|
app.requests[0].resolve(success('session'));
|
|
await setImmediate();
|
|
app.signal('SIGTERM');
|
|
assert.deepEqual(app.exits, []);
|
|
ready.resolve(app.handle);
|
|
await setImmediate();
|
|
assert.equal(app.state.stops, 1);
|
|
assert.equal(app.requests.length, 1);
|
|
assert.equal(app.process.exitCode, 0);
|
|
assert.equal(app.timers.size, 0);
|
|
assert.deepEqual(app.exits, [0]);
|
|
app.signal('SIGINT');
|
|
await setImmediate();
|
|
assert.equal(app.state.stops, 1);
|
|
assert.deepEqual(app.exits, [0]);
|
|
});
|
|
|
|
test('explicit signals await relay and all BotContext cleanup before one controlled exit', async () => {
|
|
for (const signal of ['SIGINT', 'SIGTERM']) {
|
|
const relay_stopped = deferred();
|
|
const bots_stopped = deferred();
|
|
const app = await setup({
|
|
forward: true, stop: () => relay_stopped.promise,
|
|
contexts: [{bot_id: 1}, {bot_id: 2}], stop_bot: () => bots_stopped.promise,
|
|
});
|
|
app.connect();
|
|
app.requests[0].resolve(success('session'));
|
|
await setImmediate();
|
|
app.signal(signal);
|
|
await setImmediate();
|
|
assert.deepEqual(app.stopped_bots, [1, 2]);
|
|
assert.equal(app.state.stops, 1);
|
|
assert.deepEqual(app.exits, []);
|
|
bots_stopped.resolve();
|
|
await setImmediate();
|
|
assert.deepEqual(app.exits, []);
|
|
relay_stopped.resolve();
|
|
await setImmediate();
|
|
assert.deepEqual(app.exits, [0]);
|
|
assert.ok(app.order.indexOf('relay_stopped') < app.order.indexOf('exit'));
|
|
assert.equal(app.deadlines.size, 0);
|
|
}
|
|
});
|
|
|
|
test('signal shutdown has a bounded deadline even if relay startup or Bot cleanup never settles', async () => {
|
|
const never = deferred();
|
|
const app = await setup({
|
|
forward: true, start: () => never.promise,
|
|
contexts: [{bot_id: 1}], stop_bot: () => never.promise,
|
|
});
|
|
app.connect();
|
|
app.requests[0].resolve(success('session'));
|
|
await setImmediate();
|
|
app.signal('SIGTERM');
|
|
await setImmediate();
|
|
assert.deepEqual(app.exits, []);
|
|
assert.deepEqual(app.stopped_bots, [1]);
|
|
assert.equal(app.deadlines.size, 1);
|
|
[...app.deadlines.values()][0]();
|
|
await setImmediate();
|
|
assert.deepEqual(app.exits, [1]);
|
|
assert.equal(app.deadlines.size, 0);
|
|
assert.equal(app.requests.length, 1);
|
|
});
|
|
|
|
test('signal cleanup failures wait for remaining cleanup and exit without leaking error contents', async () => {
|
|
const bots_stopped = deferred();
|
|
const app = await setup({
|
|
forward: true, stop: async () => { throw new Error('secret-token'); },
|
|
contexts: [{bot_id: 1}], stop_bot: () => bots_stopped.promise,
|
|
});
|
|
app.connect();
|
|
app.requests[0].resolve(success('session'));
|
|
await setImmediate();
|
|
app.signal('SIGINT');
|
|
await setImmediate();
|
|
assert.deepEqual(app.exits, []);
|
|
bots_stopped.resolve();
|
|
await setImmediate();
|
|
assert.deepEqual(app.exits, [1]);
|
|
assert.ok(!app.logs.join('\n').includes('secret-token'));
|
|
});
|
|
|
|
test('relay registration errors and ACK timeouts retry full registration with the same relay and Bots', async () => {
|
|
for (const failure of [486, {token: 'secret-token'}, 'timeout']) {
|
|
const app = await setup({forward: true, contexts: [{bot_id: 1}, {bot_id: 2}]});
|
|
app.connect();
|
|
app.requests[0].resolve(success('session'));
|
|
await setImmediate();
|
|
if (failure === 'timeout') app.requests[1].reject(new Error('ACK timeout secret-token'));
|
|
else app.requests[1].resolve({status: 'error', data: failure});
|
|
await setImmediate();
|
|
assert.equal(app.state.stops, 0);
|
|
assert.equal(app.socket.connected, false);
|
|
assert.equal(app.state.disconnects, 1);
|
|
assert.ok(app.state.clears >= 2);
|
|
assert.equal(app.process.exitCode, undefined);
|
|
assert.equal(app.timers.size, 1);
|
|
assert.deepEqual(app.stopped_bots, []);
|
|
assert.deepEqual(app.exits, []);
|
|
assert.ok(!app.logs.join('\n').includes('secret-token'));
|
|
app.tick();
|
|
app.connect();
|
|
assert.equal(app.requests[2].event, 'register_node');
|
|
app.requests[2].resolve(success('recovered-session'));
|
|
await setImmediate();
|
|
assert.equal(app.requests[3].event, 'register_forward');
|
|
assert.equal(app.relay_options.length, 1);
|
|
app.requests[3].resolve({status: 'success', data: 'Relay'});
|
|
await setImmediate();
|
|
assert.deepEqual(app.updates, ['plugins', 'functions', 'workflow_catalog', 'versions', 'bots',
|
|
'plugins', 'functions', 'workflow_catalog', 'versions', 'bots']);
|
|
assert.equal(app.timers.size, 0);
|
|
assert.equal(app.state.stops, 0);
|
|
assert.deepEqual(app.stopped_bots, []);
|
|
assert.equal(app.process.exitCode, undefined);
|
|
}
|
|
});
|
|
|
|
test('recurring errors keep retrying past disabled and exhausted settings without stopping relay or Bots', async () => {
|
|
const app = await setup({forward: true, contexts: [{bot_id: 5}, {bot_id: 6}],
|
|
auto_reconnect: {enabled: false, max_attempts: 1, interval: 100}});
|
|
app.connect();
|
|
app.requests[0].resolve(success('session'));
|
|
await setImmediate();
|
|
app.requests[1].resolve({status: 'success', data: 'Relay'});
|
|
await setImmediate();
|
|
app.disconnect();
|
|
for (let attempt = 0; attempt < 12; attempt++) {
|
|
app.tick();
|
|
app.dispatch('connect_error', new Error('secret-token'));
|
|
assert.equal(app.timers.size, 1);
|
|
}
|
|
await setImmediate();
|
|
assert.equal(app.state.stops, 0);
|
|
assert.equal(app.process.exitCode, undefined);
|
|
assert.equal(app.timers.size, 1);
|
|
assert.deepEqual(app.stopped_bots, []);
|
|
assert.deepEqual(app.exits, []);
|
|
assert.ok(!app.logs.join('\n').includes('secret-token'));
|
|
});
|
|
|
|
test('retry attempt count resets only after initialization and relay registration fully succeed', async () => {
|
|
const app = await setup({forward: true, auto_reconnect: {enabled: false, max_attempts: 1, interval: 100}});
|
|
for (let attempt = 0; attempt < 4; attempt++) {
|
|
app.connect();
|
|
app.requests[attempt * 2].resolve(success(`session-${attempt}`));
|
|
await setImmediate();
|
|
app.requests[attempt * 2 + 1].resolve(attempt < 3
|
|
? {status: 'error', data: 486}
|
|
: {status: 'success', data: 'Relay'});
|
|
await setImmediate();
|
|
if (attempt < 3) app.tick();
|
|
}
|
|
app.disconnect();
|
|
app.tick();
|
|
const attempts = app.logs.flatMap(message => {
|
|
const match = message.match(/正在尝试重连主服务器，第 (\d+) 次/);
|
|
return match ? [Number(match[1])] : [];
|
|
});
|
|
assert.deepEqual(attempts, [1, 2, 3, 1]);
|
|
assert.equal(app.relay_options.length, 1);
|
|
assert.equal(app.state.stops, 0);
|
|
assert.equal(app.process.exitCode, undefined);
|
|
});
|
|
|
|
test('signals cancel a pending retry and later connection callbacks cannot restart the Node', async () => {
|
|
for (const signal of ['SIGINT', 'SIGTERM']) {
|
|
const app = await setup({contexts: [{bot_id: 1}]});
|
|
app.dispatch('connect_error', new Error('secret-token'));
|
|
assert.equal(app.timers.size, 1);
|
|
app.signal(signal);
|
|
await setImmediate();
|
|
assert.equal(app.timers.size, 0);
|
|
assert.deepEqual(app.stopped_bots, [1]);
|
|
assert.deepEqual(app.exits, [0]);
|
|
app.dispatch('connect_error', new Error('secret-token'));
|
|
app.dispatch('error', new Error('secret-token'));
|
|
app.dispatch('disconnect', 'io client disconnect');
|
|
app.connect();
|
|
await setImmediate();
|
|
assert.equal(app.timers.size, 0);
|
|
assert.equal(app.requests.length, 0);
|
|
assert.equal(app.state.connects, 1);
|
|
assert.deepEqual(app.exits, [0]);
|
|
}
|
|
});
|
|
|
|
test('stale relay registration failures cannot stop a newly authenticated connection', async () => {
|
|
const app = await setup({forward: true});
|
|
app.connect();
|
|
app.requests[0].resolve(success('old-session'));
|
|
await setImmediate();
|
|
app.disconnect();
|
|
app.tick();
|
|
app.connect();
|
|
app.requests[2].resolve(success('new-session'));
|
|
await setImmediate();
|
|
app.requests[1].reject(new Error('secret-token'));
|
|
await setImmediate();
|
|
assert.equal(app.state.stops, 0);
|
|
assert.equal(app.process.exitCode, undefined);
|
|
assert.equal(app.relay_options.length, 1);
|
|
assert.ok(!app.logs.join('\n').includes('secret-token'));
|
|
});
|
|
|
|
test('obsolete initialization failures cannot stop a re-authenticated connection', async () => {
|
|
const pending = deferred();
|
|
let calls = 0;
|
|
const app = await setup({plugins: () => ++calls === 1 ? pending.promise : undefined});
|
|
app.connect();
|
|
app.requests[0].resolve(success('old-session'));
|
|
await setImmediate();
|
|
app.disconnect();
|
|
app.tick();
|
|
app.connect();
|
|
app.requests[1].resolve(success('new-session'));
|
|
await setImmediate();
|
|
pending.reject(new Error('secret-token'));
|
|
await setImmediate();
|
|
assert.equal(app.state.disconnects, 0);
|
|
assert.equal(app.process.exitCode, undefined);
|
|
assert.ok(!app.logs.join('\n').includes('secret-token'));
|
|
});
|