fix(relay): 转发回退到速度优化前的可用实现(原样直发 + 竞态帧暂存补发),仅保留异步有序落盘
- 移除会话级 socket 反压暂停/看门狗/重连重试:该改动会让握手帧在内核缓冲里卡住,表现为前端反复新建 wss、进不去战场 - 恢复:客户端帧原样直发;上游未就绪时暂存 Pending,open 后按原序补发;上游关闭即关客户端 - 保留:落盘走单条有序异步写队列(默认开启),崩溃可见性兜底
This commit is contained in:
1 parent
4005213226
commit
ad2b069659
1 file changed
+29
-117
+29
-117
@@ -55,13 +55,8 @@ const WRITE_HEX = Arg('no-hex', '0') !== '1';
|
||||
const QUIET = Arg('quiet', '0') === '1';
|
||||
const NeedDisk = DUMP_EVERY > 0 || WRITE_LOG;
|
||||
|
||||
// 转发策略(零等待、零缓存):
|
||||
// 上游就绪 → 客户端帧**原样直发**;未就绪 → **暂停读客户端 socket**(反压,字节留在内核缓冲),
|
||||
// 就绪后立刻恢复。应用层不攒帧、不回放,因此既不丢也不乱序。
|
||||
// --reconnect N 上游断开后的重连间隔(ms,默认 600)
|
||||
const RECONNECT_MS = Number(Arg('reconnect', '600'));
|
||||
// 反压看门狗:socket 被暂停时连关闭帧也读不到,若长时间无法转发就主动收摊(避免僵尸会话)
|
||||
const PAUSE_TIMEOUT_MS = Number(Arg('pause-timeout', '60000'));
|
||||
// 转发策略:上游就绪 → 客户端帧**原样直发**;未就绪(握手帧常比上游 TLS 建连更早到达)→
|
||||
// 暂存于进程内 Pending,上游一连上立即按原序补发。不重放、不改字节、不加延迟。
|
||||
|
||||
// ===== 崩溃可见性 & 稳定性 =====
|
||||
// 未捕获异常/未处理拒绝一律打印完整堆栈并**保持进程存活**,避免长对局中途"后端悄悄崩了";
|
||||
@@ -514,127 +509,44 @@ WsServer.on('connection', (Client, Req) => {
|
||||
|
||||
const Decoders = { c2s: new StreamDecoder(State, 'c2s'), s2c: new StreamDecoder(State, 's2c') };
|
||||
let Upstream = null;
|
||||
let ReconnectTimer = null;
|
||||
let PauseTimer = null;
|
||||
let Closed = false;
|
||||
let DroppedFrames = 0;
|
||||
let LastDropLogAt = 0;
|
||||
const Pending = []; // 竞态保护:暂停生效前已进入的前几帧(原序补发)
|
||||
let PendingBytes = 0;
|
||||
const MAX_RACE_FRAMES = 64;
|
||||
const Pending = []; // 上游未就绪时暂存竞态帧(握手帧常比上游 TLS 建连更早到达),一连上立即按原序补发
|
||||
|
||||
// ===== 转发路径:零等待、零缓存(靠 socket 反压保证不丢不乱序)=====
|
||||
// 上游没就绪就**暂停读客户端 socket**(字节留在内核接收缓冲里,应用层不攒帧、不回放),
|
||||
// 就绪后立刻恢复 → 客户端字节按原序直发上游。这样既不会丢帧,也不会乱序,也不存在"暂存待补发"。
|
||||
function PauseClient(Why) {
|
||||
const Sock = Client._socket;
|
||||
if (Sock && !Sock.isPaused()) {
|
||||
Sock.pause();
|
||||
State.Log('客户端读取已暂停(' + Why + ')');
|
||||
if (!PauseTimer) {
|
||||
// 注意:暂停后连客户端的关闭帧也读不到 → 必须有看门狗兜底,否则会话会僵住
|
||||
PauseTimer = setTimeout(() => {
|
||||
PauseTimer = null;
|
||||
State.Log('客户端长时间无法转发(>' + Math.round(PAUSE_TIMEOUT_MS / 1000) + 's),关闭会话以便游戏重连');
|
||||
CloseSession();
|
||||
}, PAUSE_TIMEOUT_MS);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
function ResumeClient() {
|
||||
const Sock = Client._socket;
|
||||
if (PauseTimer) { clearTimeout(PauseTimer); PauseTimer = null; }
|
||||
if (Sock && Sock.isPaused()) {
|
||||
Sock.resume();
|
||||
State.Log('上游就绪,恢复客户端转发');
|
||||
}
|
||||
}
|
||||
|
||||
function CloseSession() {
|
||||
Closed = true;
|
||||
if (PauseTimer) { clearTimeout(PauseTimer); PauseTimer = null; }
|
||||
if (ReconnectTimer) { clearTimeout(ReconnectTimer); ReconnectTimer = null; }
|
||||
if (State.Stream) State.Stream.end(); // 不落盘时 Stream 为 null,必须判空
|
||||
if (Upstream && (Upstream.readyState === Ws.OPEN || Upstream.readyState === Ws.CONNECTING)) Upstream.close();
|
||||
try { Client.terminate(); } catch (E) {}
|
||||
}
|
||||
|
||||
PauseClient('等待上游就绪'); // 立刻反压:只有能转发时才把客户端字节读进来
|
||||
|
||||
function ScheduleReconnect() {
|
||||
if (Closed || ReconnectTimer) return;
|
||||
ReconnectTimer = setTimeout(() => {
|
||||
ReconnectTimer = null;
|
||||
if (Closed) return;
|
||||
State.Log('尝试重连上游: ' + UPSTREAM);
|
||||
ConnectUpstream();
|
||||
}, RECONNECT_MS);
|
||||
}
|
||||
|
||||
function ConnectUpstream() {
|
||||
if (Closed) return;
|
||||
if (Upstream && (Upstream.readyState === Ws.OPEN || Upstream.readyState === Ws.CONNECTING)) return;
|
||||
Upstream = new Ws(UPSTREAM, { headers: UPSTREAM_HEADERS, rejectUnauthorized: false });
|
||||
Upstream.binaryType = 'nodebuffer';
|
||||
Upstream.on('open', () => {
|
||||
State.Log('上游已连接: ' + UPSTREAM + (Pending.length ? ',补发竞态帧 ' + Pending.length + ' 帧 / ' + PendingBytes + ' 字节' : ''));
|
||||
while (Pending.length) {
|
||||
const Stashed = Pending.shift();
|
||||
PendingBytes -= Stashed.length;
|
||||
Upstream.send(Stashed);
|
||||
}
|
||||
ResumeClient(); // 反压解除:客户端字节按原序继续直发
|
||||
});
|
||||
Upstream.on('message', (Data) => {
|
||||
const Buf = Buffer.from(Data);
|
||||
SaveFrame(State, 's2c', Buf);
|
||||
Decoders.s2c.Feed(Buf);
|
||||
SyncKeys(Decoders);
|
||||
if (Client.readyState === Ws.OPEN) Client.send(Buf);
|
||||
});
|
||||
Upstream.on('close', (Code) => {
|
||||
// 「持续等待」:不踢客户端;立刻反压,等重连好再继续转发(期间不攒帧、不回放)
|
||||
State.Log('上游关闭 code=' + Code + ',保持客户端连接并等待重连');
|
||||
PauseClient('上游断开,等待重连');
|
||||
ScheduleReconnect();
|
||||
});
|
||||
Upstream.on('error', (Err) => {
|
||||
State.Log('上游错误: ' + Err.message);
|
||||
PauseClient('上游错误,等待重连');
|
||||
ScheduleReconnect();
|
||||
});
|
||||
}
|
||||
|
||||
ConnectUpstream();
|
||||
// 转发:客户端帧**原样直发**;上游未就绪时暂存,open 后按原序补发(不重放、不改字节)。
|
||||
Upstream = new Ws(UPSTREAM, { headers: UPSTREAM_HEADERS, rejectUnauthorized: false });
|
||||
Upstream.binaryType = 'nodebuffer';
|
||||
Upstream.on('open', () => {
|
||||
State.Log('上游已连接: ' + UPSTREAM + ',补发暂存 ' + Pending.length + ' 帧');
|
||||
while (Pending.length) Upstream.send(Pending.shift());
|
||||
});
|
||||
Upstream.on('message', (Data) => {
|
||||
const Buf = Buffer.from(Data);
|
||||
SaveFrame(State, 's2c', Buf);
|
||||
Decoders.s2c.Feed(Buf);
|
||||
SyncKeys(Decoders);
|
||||
if (Client.readyState === Ws.OPEN) Client.send(Buf);
|
||||
});
|
||||
Upstream.on('close', (Code) => {
|
||||
State.Log('上游关闭 code=' + Code);
|
||||
if (Client.readyState === Ws.OPEN) Client.close();
|
||||
});
|
||||
Upstream.on('error', (Err) => State.Log('上游错误: ' + Err.message));
|
||||
|
||||
Client.on('message', (Data) => {
|
||||
const Buf = Buffer.from(Data);
|
||||
SaveFrame(State, 'c2s', Buf);
|
||||
Decoders.c2s.Feed(Buf);
|
||||
SyncKeys(Decoders);
|
||||
if (Upstream && Upstream.readyState === Ws.OPEN) {
|
||||
Upstream.send(Buf); // 原样直发,不等待、不缓存
|
||||
return;
|
||||
}
|
||||
// 只有"暂停生效前"已进入的帧会走到这里(正常 0~1 帧):按原序暂存,连上后立即补发
|
||||
if (Pending.length < MAX_RACE_FRAMES) {
|
||||
if (Upstream && Upstream.readyState === Ws.OPEN) Upstream.send(Buf);
|
||||
else {
|
||||
State.Log('上游尚未就绪,暂存 ' + Buf.length + ' 字节待补发');
|
||||
Pending.push(Buf);
|
||||
PendingBytes += Buf.length;
|
||||
State.Log('竞态窗口帧(暂停生效前)暂存:' + Buf.length + ' 字节(队列 ' + Pending.length + ' 帧)');
|
||||
} else {
|
||||
DroppedFrames++;
|
||||
State.Log('竞态队列已满,丢弃该帧:' + Buf.length + ' 字节 累计丢弃 ' + DroppedFrames + ' 帧');
|
||||
}
|
||||
PauseClient('上游未就绪');
|
||||
});
|
||||
|
||||
Client.on('close', (Code) => {
|
||||
State.Log(
|
||||
'客户端断开 code=' + Code + ' 收 ' + State.C2s + ' 条 / 发 ' + State.S2c + ' 条' +
|
||||
(DroppedFrames ? ' 丢弃 ' + DroppedFrames + ' 帧' : ''),
|
||||
);
|
||||
CloseSession();
|
||||
State.Log('客户端断开 code=' + Code + ' 收 ' + State.C2s + ' 条 / 发 ' + State.S2c + ' 条');
|
||||
if (State.Stream) State.Stream.end(); // 不落盘时 Stream 为 null,必须判空
|
||||
if (Upstream && Upstream.readyState === Ws.OPEN) Upstream.close();
|
||||
});
|
||||
Client.on('error', (Err) => State.Log('客户端错误: ' + Err.message));
|
||||
});
|
||||
@@ -644,6 +556,6 @@ Server.listen(PORT, () => {
|
||||
console.log(' 模式 : relay → ' + UPSTREAM);
|
||||
console.log(' 抓包目录: ' + PACKET_DIR);
|
||||
console.log(' 落盘策略: ' + (NeedDisk ? ('dump-every=' + DUMP_EVERY + ' log=' + (WRITE_LOG ? 'on' : 'off')) : '不落盘(只转发+实时解析;要抓包加 --dump-every 1 --log 1)'));
|
||||
console.log(' 转发策略: 原样直发(零等待零缓存);上游未就绪时反压暂停客户端读取,断开后每 ' + RECONNECT_MS + 'ms 重连');
|
||||
console.log(' 转发策略: 原样直发;上游未就绪时暂存竞态帧,一连上立即按原序补发');
|
||||
console.log(' 实时解析: 明文直解;加密段自动穷举密钥后实时解密;zlib 压缩帧自动解压');
|
||||
});
|
||||
Reference in new issue
Block a user