fix: 修复 WebSocket 长连接空闲断线问题
三处协同修复: 1. 前端 aibox.js 添加 25s 心跳(ping),防止 gateway 30s 空闲超时 2. gateway libhttp doWsRead 对 WS 代理连接跳过空闲超时 3. aibox.py _handle_tool_calls 启动 drain 线程实时排空输出队列, 避免 SubAgent 等长时间工具执行期间前端收不到数据
This commit is contained in:
1 parent
84a1a5e3be
commit
d680dcd058
3 files changed
+57
-14
No files matched your search
+27
-12
@@ -7133,7 +7133,32 @@ class DeepSeekChat:
|
||||
|
||||
result_raw = None
|
||||
try:
|
||||
# 启动后台线程持续排空 output_queue,将流式输出实时转发给 output_callback,
|
||||
# 避免 SubAgent 等长时间工具执行期间前端收不到数据导致 WS 空闲断线
|
||||
def _drain_queue(q=output_queue, cb=output_callback, stop_event=self._cancel_event):
|
||||
while not stop_event.is_set():
|
||||
try:
|
||||
msg_type, content = q.get(timeout=0.1)
|
||||
if cb and msg_type == "line":
|
||||
cb("line", str(content) if content is not None else "")
|
||||
except queue.Empty:
|
||||
continue
|
||||
|
||||
import threading as _th
|
||||
drain_thread = _th.Thread(target=_drain_queue, daemon=True)
|
||||
drain_thread.start()
|
||||
|
||||
result_raw = self.tool_registry.execute_tool(tool_name, **args)
|
||||
|
||||
# 工具执行完毕,排空队列中剩余内容并等待 drain 线程退出
|
||||
while not output_queue.empty():
|
||||
try:
|
||||
msg_type, content = output_queue.get_nowait()
|
||||
if output_callback and msg_type == "line":
|
||||
output_callback("line", str(content) if content is not None else "")
|
||||
except queue.Empty:
|
||||
break
|
||||
# drain 线程会在下一次 get 超时后自动退出(队列已空)
|
||||
except Exception as e:
|
||||
result_raw = f"❌ 工具 {tool_name} 执行异常: {str(e)}"
|
||||
self.logger.error(f"工具 {tool_name} 执行异常: {e}")
|
||||
@@ -7141,18 +7166,8 @@ class DeepSeekChat:
|
||||
# 安全转换并清理结果
|
||||
result = self._sanitize_tool_result(result_raw)
|
||||
|
||||
# 排空 output_queue,把流式行输出转发给 output_callback
|
||||
has_stream_output = False
|
||||
while not output_queue.empty():
|
||||
try:
|
||||
msg_type, content = output_queue.get_nowait()
|
||||
if output_callback and msg_type == "line":
|
||||
# 只转发"line"类型;"complete"是流式结束信号,不作为行输出
|
||||
# (之前complete也被转发导致工具结果在侧栏显示两次)
|
||||
output_callback("line", str(content) if content is not None else "")
|
||||
has_stream_output = True
|
||||
except queue.Empty:
|
||||
break
|
||||
# drain 线程已排空队列中的 line 输出,这里只需判断是否有流式输出
|
||||
has_stream_output = not output_queue.empty()
|
||||
|
||||
# 普通(非流式)工具没有队列输出,直接发送结果
|
||||
if not has_stream_output and output_callback and result:
|
||||
|
||||
@@ -82,8 +82,12 @@ void HttpServer::Session::doWsRead() {
|
||||
auto self = shared_from_this();
|
||||
ws_buffer_.consume(ws_buffer_.size());
|
||||
|
||||
// P1-4:WS 读取设置空闲超时,清理长期不发消息的连接
|
||||
stream_.expires_after(server_->getReadTimeout());
|
||||
// P1-4:WS 读取设置空闲超时,清理长期不发消息的连接。
|
||||
// WebSocket 代理模式(如 aibox)不设超时:后端 AI 处理可能长时间无数据,
|
||||
// 由前端心跳保活 + 后端自身超时机制管理连接生命周期。
|
||||
if (!is_ws_proxy_) {
|
||||
stream_.expires_after(server_->getReadTimeout());
|
||||
}
|
||||
|
||||
if (ssl_enabled_ && wss_stream_) {
|
||||
wss_stream_->async_read(
|
||||
@@ -329,6 +333,12 @@ void HttpServer::Session::onProxyHandshake(beast::error_code ec) {
|
||||
|
||||
proxy_connected_ = true;
|
||||
proxy_closed_ = false;
|
||||
|
||||
// 清除客户端侧流在 HTTP 升级阶段继承的读 deadline(body 阶段设置的 30s)。
|
||||
// 否则代理连接会在握手后约 30s 被强制关闭,导致长连接 WebSocket 断线。
|
||||
// 注意:后端侧 proxy_tcp_stream_ 已在 onProxyConnect 中 expires_never()。
|
||||
stream_.expires_never();
|
||||
|
||||
std::cout << "[WS-Proxy] Connected to backend " << proxy_host_ << ":" << proxy_port_
|
||||
<< proxy_path_ << std::endl;
|
||||
|
||||
|
||||
@@ -44,6 +44,21 @@ let ws = null;
|
||||
let manualClose = false; /* 手动断开则抑制自动重连 */
|
||||
let reconnectDelay = RECONNECT_MIN;
|
||||
let reconnectTimer = null;
|
||||
let heartbeatTimer = null; /* 心跳定时器,25s 发送 ping 保持 WS 连接 */
|
||||
const HEARTBEAT_INTERVAL = 25000; /* 心跳间隔(< gateway 30s 空闲超时) */
|
||||
|
||||
function startHeartbeat() {
|
||||
stopHeartbeat();
|
||||
heartbeatTimer = setInterval(() => {
|
||||
if (ws && ws.readyState === WebSocket.OPEN) {
|
||||
ws.send(JSON.stringify({ type: 'ping' }));
|
||||
}
|
||||
}, HEARTBEAT_INTERVAL);
|
||||
}
|
||||
|
||||
function stopHeartbeat() {
|
||||
if (heartbeatTimer) { clearTimeout(heartbeatTimer); heartbeatTimer = null; }
|
||||
}
|
||||
|
||||
/* 流式状态:不预创建空气泡,第一个 chunk 到来才挂载(保证思考区必然在正式回答之前) */
|
||||
let aiBubble = null, aiText = '', aiActive = false;
|
||||
@@ -204,6 +219,7 @@ function connect() {
|
||||
setStatus('connected', '已连接');
|
||||
connBtn.textContent = '断开';
|
||||
connBtn.classList.add('on');
|
||||
startHeartbeat();
|
||||
};
|
||||
ws.onmessage = (ev) => {
|
||||
let data;
|
||||
@@ -212,6 +228,7 @@ function connect() {
|
||||
routeMessage(data);
|
||||
};
|
||||
ws.onclose = () => {
|
||||
stopHeartbeat();
|
||||
setStatus('disconnected', '已断开');
|
||||
connBtn.textContent = '连接';
|
||||
connBtn.classList.remove('on');
|
||||
@@ -235,6 +252,7 @@ function scheduleReconnect() {
|
||||
function disconnect(manual) {
|
||||
manualClose = manual !== false;
|
||||
clearTimeout(reconnectTimer);
|
||||
stopHeartbeat();
|
||||
if (ws) { ws.close(); ws = null; }
|
||||
setStatus('disconnected', manualClose ? '已手动断开' : '已断开');
|
||||
connBtn.textContent = '连接';
|
||||
|
||||
Reference in new issue
Block a user