2508932142/node_server
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
main
node_server/tests/forward_manager.test.mjs
T
2508932142 6c69afb38a feat: add world voxel snapshot forwarding 
Collect chunk voxel data from the bot world and stream snapshot,
chunk, and unwatch messages through the forward protocol.
2026-10-23 10:44:37 +08:00

439 lines
23 KiB
JavaScript
Raw Permalink Blame History

import assert from 'node:assert/strict';
import {EventEmitter} from 'node:events';
import {createRequire} from 'node:module';
import {randomUUID} from 'node:crypto';
import {fileURLToPath} from 'node:url';
import {readFile} from 'node:fs/promises';
import vm from 'node:vm';
import test from 'node:test';
import {build, transformSync} from 'esbuild';
const require = createRequire(import.meta.url);
const root = fileURLToPath(new URL('../', import.meta.url));
const built = await build({
stdin: {contents: `export * from './bot/forward/manager.ts';
export {handler} from './event_handler/bot/forward_bot.ts';`, resolveDir: root, loader: 'ts'},
bundle: true, platform: 'node', format: 'cjs',
packages: 'external', write: false,
plugins: [{name: 'stream-stub', setup(build) {
build.onResolve({filter: /^\.\/stream\.ts$/}, () => ({path: 'stream', namespace: 'mock'}));
build.onLoad({filter: /.*/, namespace: 'mock'}, () => ({contents: 'export class BotStream {}', loader: 'js'}));
}}],
});
const module = {exports: {}};
vm.runInNewContext(built.outputFiles[0].text, {module, exports: module.exports, require,
Buffer, URL, setTimeout, clearTimeout, console});
const {NodeForwardManager} = module.exports;
function command(overrides = {}) {
return {forward_id: 0, forward_session_id: randomUUID(), node_session_id: 'node-session', node_id: 1, bot_id: 2,
url: 'http://127.0.0.1:1234', path: '/api/forward_server', namespace: '/', token: randomUUID(),
expires_at: Date.now() + 30_000, deadline_at: Date.now() + 500, ...overrides};
}
function setup() {
const sockets = [];
const streams = [];
const files = [];
const hooks = new Map();
const initial = command();
let active = initial;
const manager = new NodeForwardManager(1, 'node-session', {
socket_factory(url, options) {
const socket = new EventEmitter();
socket.sent = [];
socket.url = url;
socket.options = options;
socket.auth = options.auth;
socket.connected = false;
socket.volatile = {emit: (...args) => socket.sent.push(args)};
socket.io = {engine: {writeBuffer: []}};
socket.connect = () => {
socket.connected = true;
queueMicrotask(() => socket.emit('connect'));
};
socket.disconnect = () => {
socket.connected = false;
socket.emit('disconnect');
};
socket.timeout = () => ({emit(event, payload, ack) {
socket.sent.push([event, payload]);
const respond = data => ack(null, {status: 'success', data});
if (hooks.has(event)) return hooks.get(event)({socket, payload, respond, ack});
if (event === 'forward_node_ready') return queueMicrotask(() => respond({node_ready: true}));
const {forward_id, forward_session_id, node_session_id, node_id, bot_id, expires_at} = active;
queueMicrotask(() => respond({forward_id, forward_session_id, node_session_id, node_id, bot_id,
role: 'node', node_ready: event === 'reauth', expires_at}));
}});
sockets.push(socket);
return socket;
},
stream_factory(bot_id) {
const listeners = new Set();
const stream = {bot_id, disposed: false, chats: 0,
snapshot: () => ({bot_id}), logs: () => ({records: []}),
get_context_info: () => [{name: 'health', i18n: {lang: 'en', name: 'Health', description: ''}, value: 20}],
chat: () => { stream.chats++; return {accepted: true}; },
subscribe: listener => { listeners.add(listener); return () => listeners.delete(listener); },
resync() {}, dispose() { stream.disposed = true; },
event: event => { for (const listener of listeners) listener(event); },
get listeners() { return listeners.size; },
};
streams.push(stream);
return stream;
},
file_factory(bot_id) {
const service = {bot_id, disposed: false,
list: async () => ({listing_id: randomUUID(), expires_at: Date.now() + 300_000, partial: false, children: []}),
read: async ({id, offset = 0}) => ({id, name: 'empty', encoding: 'base64', offset,
next_offset: offset, total_bytes: offset, eof: true, data: ''}),
dispose() { service.disposed = true; },
};
files.push(service);
return service;
},
});
return {manager, sockets, streams, files, hooks, command: active,
ensure(value = initial) { active = value; return manager.ensure(value); }};
}
test('lifecycle updates use reliable bot_event even while volatile logs are saturated', async t => {
const f = setup(); t.after(() => f.manager.stop());
await f.ensure();
const socket = f.sockets[0], stream = f.streams[0], received = [];
socket.on('bot_event', event => received.push(event));
socket.volatile = {emit() {}};
const base = {bot_id: 2, context_generation: randomUUID(), stream_id: randomUUID(), emitted_at: Date.now()};
for (let seq = 1; seq <= 160; seq++) stream.event({...base, seq, name: 'log', payload: {records: []}});
for (const [index, status] of ['online', 'disconnecting', 'offline'].entries()) {
stream.event({...base, seq: 161 + index, name: 'status_update', payload: {status, context_exists: true}});
}
assert.deepEqual(received.map(event => event.payload.status), ['online', 'disconnecting', 'offline']);
assert.ok(received.every(event => event.forward_session_id === f.command.forward_session_id));
stream.event({...base, seq: 165, name: 'message', payload: {text: 'not a lifecycle event'}});
assert.equal(received.length, 3);
socket.io.engine.writeBuffer.length = 64;
stream.event({...base, seq: 166, name: 'status_update', payload: {status: 'offline', context_exists: true}});
assert.equal(socket.connected, false);
assert.equal(stream.listeners, 0);
})
const tick = () => new Promise(resolve => setImmediate(resolve));
const code = response => response.data.code;
test('file RPCs work without a bot, clamp ID expiry and invalidate services on transport replacement', async t => {
const app = setup(); t.after(() => app.manager.stop());
await app.ensure();
const rpc = (socket, method, payload = {}, deadline_at = Date.now() + 1000) => new Promise(resolve =>
socket.emit('relay_request', {request_id: randomUUID(), correlation_id: randomUUID(), method, payload,
forward_session_id: app.command.forward_session_id, deadline_at}, resolve));
const first = app.sockets[0];
const listed = await rpc(first, 'get_data_file_list');
assert.equal(listed.status, 'success');
assert.equal(listed.data.expires_at, app.command.expires_at);
assert.equal(app.files.length, 1);
assert.equal(app.files[0].bot_id, app.command.bot_id);
const id = randomUUID();
assert.equal((await rpc(first, 'get_file_data', {id})).data.id, id);
assert.equal(code(await rpc(first, 'get_file_data', {id, length: 262145})), 'INVALID_REQUEST');
assert.equal(code(await rpc(first, 'get_file_data', {id, offset: -1})), 'INVALID_REQUEST');
let release;
app.files[0].read = () => new Promise(resolve => {release = resolve;});
const pending = rpc(first, 'get_file_data', {id});
first.disconnect();
assert.equal(app.files[0].disposed, true);
release({id});
assert.equal(code(await pending), 'SESSION_STALE');
assert.equal((await app.ensure({...app.command, token: randomUUID()})).status, 'success');
assert.equal((await rpc(app.sockets[1], 'get_data_file_list')).status, 'success');
assert.equal(app.files.length, 2);
let late;
app.files[1].list = () => new Promise(resolve => {late = resolve;});
const timed = rpc(app.sockets[1], 'get_data_file_list', {}, Date.now() + 10);
await new Promise(resolve => setTimeout(resolve, 15));
late({listing_id: randomUUID(), expires_at: Date.now() + 1000, partial: false, children: []});
assert.equal(code(await timed), 'TIMEOUT');
app.manager.stop();
assert.equal(app.files[1].disposed, true);
});
test('ready ACK waits for ready confirmation, with one RPC handler and log observer across renewals', async t => {
const app = setup();
t.after(() => app.manager.stop());
let release;
app.hooks.set('forward_node_ready', ({respond}) => { release = respond; });
let completed = false;
const result = app.ensure().then(value => { completed = true; return value; });
await tick();
assert.equal(completed, false);
assert.equal(app.sockets[0].listenerCount('relay_request'), 1);
assert.equal(app.streams[0].listeners, 1);
release({node_ready: true});
assert.equal((await result).status, 'success');
app.hooks.delete('forward_node_ready');
const next = {...app.command, token: randomUUID()};
assert.equal((await app.ensure(next)).status, 'success');
assert.equal(app.sockets.length, 1);
assert.equal(app.streams[0].listeners, 1);
assert.equal(app.sockets[0].sent.filter(([event]) => event === 'reauth').length, 1);
});
test('absolute deadline cancels stalled readiness, closes channel and ignores late ready ACK', async t => {
const app = setup();
t.after(() => app.manager.stop());
let release;
app.hooks.set('forward_node_ready', ({respond}) => { release = respond; });
const request = {...app.command, deadline_at: Date.now() + 35};
const response = await app.ensure(request);
assert.equal(code(response), 'TIMEOUT');
assert.equal(app.sockets[0].connected, false);
assert.equal(app.sockets[0].listenerCount('relay_request'), 0);
assert.equal(app.streams[0].disposed, true);
release({node_ready: true});
await tick();
assert.equal(app.sockets[0].connected, false);
});
test('session scope and whitelist validation reject stale node, wrong bot scope, and extra fields', async t => {
const app = setup();
t.after(() => app.manager.stop());
assert.equal(code(await app.ensure({...app.command, node_session_id: 'old'})), 'SESSION_STALE');
assert.equal(code(await app.ensure({...app.command, node_id: 3})), 'FORBIDDEN');
assert.equal(code(await app.ensure({...app.command, extra: true})), 'INVALID_REQUEST');
assert.equal((await app.ensure()).status, 'success');
assert.equal(code(await app.ensure({...app.command, bot_id: 3})), 'SESSION_STALE');
assert.equal(app.sockets.length, 1);
});
test('serial ensure ignores expired queued work without closing a healthy current operation', async t => {
const app = setup();
t.after(() => app.manager.stop());
let release;
app.hooks.set('forward_node_ready', ({respond}) => { release = respond; });
const first = app.ensure();
await tick();
const second = app.ensure({...app.command, token: randomUUID(), deadline_at: Date.now() + 20});
assert.equal(code(await second), 'TIMEOUT');
assert.equal(app.sockets[0].connected, true);
release({node_ready: true});
assert.equal((await first).status, 'success');
await tick();
assert.equal(app.sockets.length, 1);
assert.equal(app.sockets[0].connected, true);
});
test('closed session cannot revive, old callbacks cannot execute chat, and manager stop removes observers', async () => {
const app = setup();
assert.equal((await app.ensure()).status, 'success');
const socket = app.sockets[0];
const stale_rpc = socket.listeners('relay_request')[0];
socket.emit('forward_session_closed', {forward_session_id: app.command.forward_session_id,
reason: 'revoked', code: 'CANCELLED'});
assert.equal(app.streams[0].disposed, true);
assert.equal(socket.connected, false);
assert.equal(code(await app.ensure({...app.command, token: randomUUID()})), 'SESSION_STALE');
let response;
stale_rpc({request_id: randomUUID(), correlation_id: randomUUID(), method: 'bot.chat', payload: {message: 'hello'},
forward_session_id: app.command.forward_session_id, deadline_at: Date.now() + 1000}, value => { response = value; });
assert.equal(code(response), 'SESSION_STALE');
assert.equal(app.streams[0].chats, 0);
app.manager.stop();
assert.equal(code(await app.ensure(command())), 'SESSION_STALE');
});
test('RPC checks deadlines and deduplicates correlation IDs, while allowing grant request ID reuse', async t => {
const app = setup();
t.after(() => app.manager.stop());
await app.ensure();
const base = {request_id: randomUUID(), correlation_id: randomUUID(), method: 'bot.chat', payload: {message: 'hello'},
forward_session_id: app.command.forward_session_id, deadline_at: Date.now() + 1000};
const rpc = request => new Promise(resolve => app.sockets[0].emit('relay_request', request, resolve));
assert.equal((await rpc(base)).status, 'success');
assert.equal(code(await rpc(base)), 'CANCELLED');
assert.equal((await rpc({...base, correlation_id: randomUUID()})).status, 'success');
assert.equal(app.streams[0].chats, 2);
assert.equal(code(await rpc({...base, correlation_id: randomUUID(), deadline_at: Date.now() - 1})), 'TIMEOUT');
assert.equal(app.streams[0].chats, 2);
assert.equal(code(await rpc({...base, method: 'bot.subscribe', payload: {events: []}})), 'INVALID_REQUEST');
const info = await rpc({...base, correlation_id: randomUUID(), method: 'get_context_info', payload: {}});
assert.equal(info.status, 'success');
assert.equal(info.data[0].value, 20);
});
test('transport recovery requires a fresh token; ordinary disconnect never reconnects automatically', async t => {
const app = setup();
t.after(() => app.manager.stop());
await app.ensure();
app.sockets[0].disconnect();
assert.equal(app.sockets[0].options.reconnection, false);
assert.equal(code(await app.ensure(app.command)), 'TOKEN_EXPIRED');
assert.equal(app.sockets.length, 1);
assert.equal((await app.ensure({...app.command, token: randomUUID()})).status, 'success');
assert.equal(app.sockets.length, 2);
assert.equal(app.streams.length, 1);
assert.equal(app.streams[0].listeners, 1);
});
test('wrong-scope join response closes the channel and sanitized ACK never exposes raw relay details', async t => {
const app = setup();
t.after(() => app.manager.stop());
app.hooks.set('join_room', ({respond}) => respond({node_id: 999, token: 'secret', stack: 'private'}));
const response = await app.ensure();
assert.equal(code(response), 'SESSION_STALE');
assert.equal(app.sockets[0].connected, false);
assert.doesNotMatch(JSON.stringify(response), /secret|private|stack/);
});
test('many historical closed sessions do not consume the active-session budget', async t => {
const app = setup();
t.after(() => app.manager.stop());
for (let index = 0; index < 505; index++) {
const request = command({deadline_at: Date.now() + 2000});
assert.equal((await app.ensure(request)).status, 'success');
app.sockets.at(-1).emit('forward_session_closed', {forward_session_id: request.forward_session_id,
reason: 'revoked', code: 'CANCELLED'});
}
assert.equal(app.streams.filter(stream => !stream.disposed).length, 0);
assert.equal((await app.ensure(command())).status, 'success');
});
test('lease expiry retires data channel and queue capacity stays bounded under stalled work', async t => {
const app = setup();
t.after(() => app.manager.stop());
const request = {...app.command, expires_at: Date.now() + 40};
assert.equal((await app.ensure(request)).status, 'success');
await new Promise(resolve => setTimeout(resolve, 50));
assert.equal(app.sockets[0].connected, false);
assert.equal(app.streams[0].disposed, true);
assert.equal(code(await app.ensure({...request, expires_at: Date.now() + 1000, token: randomUUID()})), 'SESSION_STALE');
const stalled = setup();
t.after(() => stalled.manager.stop());
stalled.hooks.set('forward_node_ready', () => {});
const first = stalled.ensure();
await tick();
const queued = Array.from({length: 15}, () => stalled.ensure({...stalled.command, token: randomUUID()}));
assert.equal(code(await stalled.ensure({...stalled.command, token: randomUUID()})), 'RATE_LIMITED');
stalled.manager.stop();
const replies = await Promise.all([first, ...queued]);
assert.ok(replies.every(reply => reply.status === 'error'));
assert.equal(stalled.sockets.length, 1);
});
test('lifecycle events are forwarded while ready and congestion closes the stale transport', async t => {
const app = setup();
t.after(() => app.manager.stop());
await app.ensure();
const socket = app.sockets[0];
const stream = app.streams[0];
let resyncs = 0;
stream.resync = () => { resyncs++; };
const event = {name: 'status_update', seq: 1, payload: {status: 'offline', context_exists: false}};
let lifecycle = 0;
socket.on('bot_event', () => lifecycle++);
stream.event(event);
assert.equal(lifecycle, 1);
assert.equal(socket.sent.filter(([name]) => name === 'bot_log').length, 0);
stream.event({name: 'log', seq: 2, payload: {records: []}});
assert.equal(socket.sent.filter(([name]) => name === 'bot_log').length, 1);
socket.emit('relay_status', {forward_session_id: app.command.forward_session_id, node_ready: true, reason: 'resync_required'});
assert.equal(resyncs, 1);
socket.io.engine.writeBuffer.length = 64;
stream.event({...event, seq: 2});
stream.event({name: 'log', seq: 3, payload: {records: []}});
assert.equal(socket.sent.filter(([name]) => name === 'bot_log').length, 1);
assert.equal(lifecycle, 1);
assert.equal(socket.connected, false);
socket.emit('relay_status', {forward_session_id: app.command.forward_session_id, node_ready: true, reason: 'resync_required'});
assert.equal(resyncs, 1);
socket.disconnect();
stream.event({...event, seq: 3});
stream.event({name: 'log', seq: 4, payload: {records: []}});
assert.equal(socket.sent.filter(([name]) => name === 'bot_log').length, 1);
assert.equal(lifecycle, 1);
assert.equal(stream.listeners, 0);
});
test('info updates use the data envelope and share push congestion and rate controls with logs', async t => {
const app = setup(); t.after(() => app.manager.stop());
let ready;
app.hooks.set('forward_node_ready', ({respond}) => {ready = respond;});
const pending = app.ensure(); await tick();
const stream = app.streams[0]; const socket = app.sockets[0];
const event = {bot_id: 2, context_generation: randomUUID(), stream_id: randomUUID(), emitted_at: Date.now(),
seq: 1, name: 'info_update', payload: {name: 'health', value: 18}};
stream.event(event);
assert.equal(socket.sent.some(([name]) => name === 'info_update'), false);
ready({node_ready: true}); await pending;
stream.event(event);
const packet = socket.sent.find(([name]) => name === 'info_update')[1];
assert.deepEqual(Object.keys(packet).sort(), ['bot_id', 'context_generation', 'data', 'emitted_at', 'forward_session_id', 'seq', 'stream_id']);
assert.equal(packet.data.value, 18);
assert.equal(packet.forward_session_id, app.command.forward_session_id);
socket.io.engine.writeBuffer.length = 64; stream.event({...event, seq: 2});
assert.equal(socket.sent.filter(([name]) => name === 'info_update').length, 1);
socket.io.engine.writeBuffer.length = 0;
for (let seq = 3; seq < 203; seq++) stream.event({...event, seq, name: seq % 2 ? 'log' : 'info_update'});
assert.equal(socket.sent.filter(([name]) => ['info_update', 'bot_log'].includes(name)).length, 149);
socket.disconnect(); stream.event({...event, seq: 204});
assert.equal(stream.listeners, 0);
assert.equal(socket.sent.filter(([name]) => ['info_update', 'bot_log'].includes(name)).length, 149);
});
test('plugin lifecycle updates survive a temporarily unwritable transport and retain congestion bounds', async t => {
const app = setup(); t.after(() => app.manager.stop());
await app.ensure();
const socket = app.sockets[0], stream = app.streams[0];
const received = [];
socket.on('info_update', update => received.push(update));
// Match Socket.IO volatile behavior during an outstanding WebSocket write.
socket.volatile = {emit() {}};
const event = {bot_id: 2, context_generation: randomUUID(), stream_id: randomUUID(), emitted_at: Date.now(),
name: 'info_update', payload: {name: 'plugin_states', value: {available: true, plugins: []}}};
for (const [index, lifecycle] of ['initializing', 'enabled'].entries()) {
stream.event({...event, seq: index + 1, payload: {...event.payload, value: {available: true,
plugins: [{name: 'auto_reconnect', path: 'runtime', lifecycle, enabled: lifecycle === 'enabled', states: {attempts: 0}}]}}});
}
assert.deepEqual(received.map(update => update.data.value.plugins[0].lifecycle), ['initializing', 'enabled']);
assert.deepEqual(received.map(update => update.seq), [1, 2]);
assert.ok(received.every(update => update.forward_session_id === app.command.forward_session_id));
assert.ok(received.every(update => !('name' in update) && !('payload' in update)));
stream.event({...event, seq: 3, payload: {name: 'health', value: 20}});
assert.equal(received.length, 2);
socket.io.engine.writeBuffer.length = 64;
stream.event({...event, seq: 4});
assert.equal(received.length, 2);
socket.io.engine.writeBuffer.length = 0;
for (let seq = 5; seq <= 155; seq++) stream.event({...event, seq});
assert.equal(received.length, 148);
socket.disconnect(); stream.event({...event, seq: 156});
assert.equal(received.length, 148);
assert.equal(stream.listeners, 0);
});
test('forward_bot handler requires an ACK and returns structured authorization errors', async () => {
const {handler} = module.exports;
const socket = {connected: true, emit() { assert.fail('Unexpected non-ACK response'); }};
await handler(socket, command(), undefined);
const replies = [];
await handler(socket, command(), reply => replies.push(reply));
assert.equal(replies.length, 1);
assert.equal(code(replies[0]), 'UNAUTHORIZED');
socket.connected = false;
await handler(socket, command(), reply => replies.push(reply));
assert.equal(code(replies[1]), 'NODE_OFFLINE');
});
test('dispatcher delivers forward_bot while hosted relay control events remain on their own handlers', async () => {
const source = await readFile(new URL('../event_handler/init.ts', import.meta.url), 'utf8');
const dispatch = transformSync(source.slice(source.indexOf('export async function event_handler'))
.replace('export async function', 'async function'), {loader: 'ts'}).code;
const calls = [];
const map = new Map([['forward_bot', (...args) => calls.push(args)]]);
const context = {handler_map: map, console_print: {log() {}, warning() {}, error() {}}, socket: {}, payload: command()};
await vm.runInNewContext(`${dispatch}\nevent_handler(socket, 'forward_bot', payload, () => {});`, context);
assert.equal(calls.length, 1);
await vm.runInNewContext("event_handler(socket, 'forward_open_session', payload, () => {});", context);
assert.equal(calls.length, 1);
});
Reference in New Issue View Git Blame Copy Permalink