From 341b130d9b724886202a904594c8860380af7c23 Mon Sep 17 00:00:00 2001 From: Nexus Agent Date: Wed, 10 Jun 2026 21:54:50 +0800 Subject: [PATCH] =?UTF-8?q?fix(rdp):=20guacd=20=E5=90=8E=E5=8F=B0=E8=AF=BB?= =?UTF-8?q?=E5=8F=96=E9=98=9F=E5=88=97=20+=20=E9=9A=A7=E9=81=93=E8=B6=85?= =?UTF-8?q?=E6=97=B6=E4=B8=8E=E5=B0=BA=E5=AF=B8?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 独立 guacd reader 避免 WS 竞争丢数据；ping 回显与 sync 已生产验证；前端 90s 超时与 OPEN 后发尺寸。 Co-authored-by: Cursor --- .../2026-06-10-rdp-guacd-reader-queue.md | 9 ++ .../2026-06-10-rdp-ws-relay-single-loop.md | 20 +++ frontend/src/composables/rdp/useRdpSession.ts | 20 ++- server/infrastructure/guacamole/ws_tunnel.py | 118 ++++++++++++------ 4 files changed, 126 insertions(+), 41 deletions(-) create mode 100644 docs/changelog/2026-06-10-rdp-guacd-reader-queue.md create mode 100644 docs/changelog/2026-06-10-rdp-ws-relay-single-loop.md diff --git a/docs/changelog/2026-06-10-rdp-guacd-reader-queue.md b/docs/changelog/2026-06-10-rdp-guacd-reader-queue.md new file mode 100644 index 00000000..7296933c --- /dev/null +++ b/docs/changelog/2026-06-10-rdp-guacd-reader-queue.md @@ -0,0 +1,9 @@ +# 2026-06-10 — RDP guacd 后台读取队列 + +## 摘要 + +WebSocket 中继改为独立 guacd 读取任务 + 队列，避免 cancel 读任务丢 TCP 数据；前端隧道 OPEN 后发送尺寸。 + +## 验证 + +生产容器内 ping 回显 + 20s 内收到 sync。 diff --git a/docs/changelog/2026-06-10-rdp-ws-relay-single-loop.md b/docs/changelog/2026-06-10-rdp-ws-relay-single-loop.md new file mode 100644 index 00000000..5393f3ff --- /dev/null +++ b/docs/changelog/2026-06-10-rdp-ws-relay-single-loop.md @@ -0,0 +1,20 @@ +# 2026-06-10 — RDP WebSocket 单循环中继（修复 Server timeout） + +## 摘要 + +修复 Guacamole 客户端报 `Server timeout.`：双协程并发 `send/receive` 违反 Starlette WebSocket 约束，隧道 ping 无法回显；改为单循环中继，并按指令拆分转发。 + +## 动机 + +客户端 `receiveTimeout` 默认 15s，期间必须收到服务端数据（含 ping 回显）。并发 `ws_to_guacd` / `guacd_to_ws` 会导致 ping 丢失。 + +## 涉及文件 + +- `server/infrastructure/guacamole/ws_tunnel.py` +- `server/infrastructure/guacamole/protocol.py` — `iter_raw_instructions` +- `frontend/src/composables/rdp/useRdpSession.ts` — `receiveTimeout=90s` +- `tests/test_guacamole_protocol.py` + +## 验证 + +生产：3389 连接不再 15s 超时；guacd 日志有 RDPDR 连接。 diff --git a/frontend/src/composables/rdp/useRdpSession.ts b/frontend/src/composables/rdp/useRdpSession.ts index 16a6c3e4..ca766da1 100644 --- a/frontend/src/composables/rdp/useRdpSession.ts +++ b/frontend/src/composables/rdp/useRdpSession.ts @@ -31,7 +31,11 @@ function buildConnectString(info: RdpConnectInfo): string { } function tunnelErrorMessage(st: { code?: number; message?: string }): string { - return st?.message || `RDP 隧道错误 (${st?.code ?? 'unknown'})` + const msg = (st?.message || '').trim() + if (/server timeout/i.test(msg)) { + return '连接超时：服务端未收到远程桌面数据，请稍后重试' + } + return msg || `RDP 隧道错误 (${st?.code ?? 'unknown'})` } export function useRdpSession() { @@ -116,6 +120,9 @@ export function useRdpSession() { `/ws/rdp/hosts/${hostId}/tunnel/${encodeURIComponent(auth.token)}`, ) const rdpTunnel = new Guacamole.WebSocketTunnel(wsUrl) + // RDP handshake to Azure can exceed the default 15s; pings need timely echo from server. + rdpTunnel.receiveTimeout = 90_000 + rdpTunnel.unstableThreshold = 15_000 tunnel = rdpTunnel const rdpClient = new Guacamole.Client(rdpTunnel) client = rdpClient @@ -130,7 +137,9 @@ export function useRdpSession() { rdpClient.onerror = (st) => fail(tunnelErrorMessage(st)) rdpClient.onstatechange = (state: number) => { - if (state === Guacamole.Client.STATE_CONNECTED) { + if (state === Guacamole.Client.STATE_WAITING) { + status.value = 'connecting' + } else if (state === Guacamole.Client.STATE_CONNECTED) { status.value = 'connected' sendDisplaySize() } else if (state === Guacamole.Client.STATE_DISCONNECTED) { @@ -138,6 +147,13 @@ export function useRdpSession() { } } + rdpTunnel.onstatechange = (state: number) => { + // Guacamole.Tunnel.State.OPEN === 1 + if (state === 1 && status.value === 'connecting') { + sendDisplaySize() + } + } + const display = rdpClient.getDisplay() display.onresize = () => sendDisplaySize() diff --git a/server/infrastructure/guacamole/ws_tunnel.py b/server/infrastructure/guacamole/ws_tunnel.py index 058badad..e4ddde0c 100644 --- a/server/infrastructure/guacamole/ws_tunnel.py +++ b/server/infrastructure/guacamole/ws_tunnel.py @@ -3,6 +3,7 @@ from __future__ import annotations import asyncio +import contextlib import logging from fastapi import WebSocket, WebSocketDisconnect @@ -11,12 +12,83 @@ from server.infrastructure.guacamole.protocol import ( GuacdStream, guacd_rdp_handshake, is_internal_instruction, + iter_raw_instructions, tunnel_open_instruction, ) logger = logging.getLogger("nexus.guacamole.ws_tunnel") +async def _guacd_reader_loop(stream: GuacdStream, out: asyncio.Queue[bytes | None]) -> None: + """Continuously read complete guacd instructions into a queue (never cancel mid-read).""" + try: + while True: + batch = await stream.read_raw_batch() + if not batch: + break + await out.put(batch) + except ConnectionError: + pass + except Exception as exc: + logger.warning("guacd reader ended: %s", exc) + finally: + await out.put(None) + + +async def _relay_ws_guacd( + websocket: WebSocket, + stream: GuacdStream, + writer: asyncio.StreamWriter, +) -> None: + """Single WS consumer; guacd reads run in a dedicated task (Starlette-safe).""" + guacd_queue: asyncio.Queue[bytes | None] = asyncio.Queue(maxsize=512) + reader_task = asyncio.create_task(_guacd_reader_loop(stream, guacd_queue)) + ws_lock = asyncio.Lock() + + async def ws_send(text: str) -> None: + async with ws_lock: + await websocket.send_text(text) + + async def forward_client_message(data: str) -> None: + for inst in iter_raw_instructions(data): + if is_internal_instruction(inst): + await ws_send(inst) + else: + writer.write(inst.encode("utf-8")) + await writer.drain() + + try: + while True: + ws_recv_task = asyncio.create_task(websocket.receive_text()) + guacd_recv_task = asyncio.create_task(guacd_queue.get()) + done, pending = await asyncio.wait( + {ws_recv_task, guacd_recv_task}, + return_when=asyncio.FIRST_COMPLETED, + ) + for task in pending: + task.cancel() + with contextlib.suppress(asyncio.CancelledError): + await task + + if ws_recv_task in done: + exc = ws_recv_task.exception() + if exc is not None: + if isinstance(exc, WebSocketDisconnect): + return + raise exc + await forward_client_message(ws_recv_task.result()) + + if guacd_recv_task in done: + batch = guacd_recv_task.result() + if batch is None: + return + await ws_send(batch.decode("utf-8", errors="replace")) + finally: + reader_task.cancel() + with contextlib.suppress(asyncio.CancelledError): + await reader_task + + async def run_guacamole_rdp_tunnel( websocket: WebSocket, guacd_host: str, @@ -38,50 +110,18 @@ async def run_guacamole_rdp_tunnel( except Exception as exc: logger.warning("guacd handshake failed: %s", exc) writer.close() - try: + with contextlib.suppress(Exception): await writer.wait_closed() - except Exception: - pass await websocket.close(code=1011, reason=str(exc)[:120]) return - async def ws_to_guacd() -> None: - try: - while True: - data = await websocket.receive_text() - if is_internal_instruction(data): - # Tunnel stability pings use empty opcode — echo, do not forward to guacd. - await websocket.send_text(data) - continue - writer.write(data.encode("utf-8")) - await writer.drain() - except WebSocketDisconnect: - pass - except Exception as exc: - logger.debug("RDP WS→guacd ended: %s", exc) - - async def guacd_to_ws() -> None: - try: - while True: - # Must not split mid-instruction — guacamole-common-js parses per WS message. - batch = await stream.read_raw_batch() - if not batch: - break - await websocket.send_text(batch.decode("utf-8", errors="replace")) - except ConnectionError: - pass - except Exception as exc: - logger.debug("RDP guacd→WS ended: %s", exc) - - ws_task = asyncio.create_task(ws_to_guacd()) - guacd_task = asyncio.create_task(guacd_to_ws()) try: - await asyncio.wait([ws_task, guacd_task], return_when=asyncio.FIRST_COMPLETED) + await _relay_ws_guacd(websocket, stream, writer) + except WebSocketDisconnect: + pass + except Exception as exc: + logger.warning("RDP relay ended: %s", exc) finally: - for task in (ws_task, guacd_task): - task.cancel() writer.close() - try: + with contextlib.suppress(Exception): await writer.wait_closed() - except Exception: - pass