diff --git a/3DTank/tools/wss_server.js b/3DTank/tools/wss_server.js index 1481dcd..ace5de8 100644 --- a/3DTank/tools/wss_server.js +++ b/3DTank/tools/wss_server.js @@ -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 压缩帧自动解压'); });