From 64facc5701d28092e9eb99c9646b3dc91382c6a3 Mon Sep 17 00:00:00 2001 From: yuetsh <517252939@qq.com> Date: Thu, 27 Aug 2026 05:07:26 -0600 Subject: [PATCH] =?UTF-8?q?fix(WebSocket):=20=E6=96=AD=E7=BA=BF=E4=B8=8D?= =?UTF-8?q?=E9=87=8D=E8=BF=9E=E3=80=81=E8=AF=84=E5=88=86=E7=BB=93=E6=9E=9C?= =?UTF-8?q?=E4=BC=9A=E4=B8=A2=E3=80=81=E7=99=BB=E5=87=BA=E5=90=8E=E8=BF=9E?= =?UTF-8?q?=E6=8E=A5=E8=BF=98=E6=B4=BB=E7=9D=80?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 排查 WS 这一块时发现的一批问题，多数是「机制写了但从没生效过」。 ## 重连 `disconnect()` 里 `enableAutoReconnect = false`，而 `connect()` 从不改回 true —— 登出再登录后，这条连接就永远失去了自动重连能力（configUpdate 那条 watch 上尤其 明显）。改成用 `closedByUser` 表达「用户主动断开」的意图，和 `enableAutoReconnect` 这个**配置**分开。 `scheduleDisconnect` 的回调里断完紧接着一句 `enableAutoReconnect = true`，而 close 是异步的 —— 等 onclose 跑到时标志已经翻回来了，1 秒后又自动连上。那个 「15 分钟空闲省资源」从来没真正断开过。现在只断开，不做事后翻转。 重连的 setTimeout 没存句柄，组件卸载后照样触发 `connect()`，在已销毁的组件上 又建一条连接。现在 `disconnect()` 里 clearTimeout。 退避从「线性 ×5 次」改成「指数 + 抖动、30 秒封顶、次数不封顶」。原来 1+2+3+4+5 只有 15 秒，后端 deploy 重启一次就超了，之后这条连接死到用户刷新为止。抖动是 为了避免一个班几十台机器在同一毫秒一起冲回刚起来的后端。另挂 online / visibilitychange，网络恢复或切回标签页立刻重连，不必等退避走完。 所有 socket 回调改成闭包住局部 ws 并在入口 `if (ws !== this.ws) return`，旧连接 迟到的 onclose 不再污染新连接的状态 —— 也是让 `disconnect()` 能被 onclose 识别 出来的关键。 ## 订阅重放 `pendingSubmissionId` 一发送成功就清空，它只解决了「还没连上就 subscribe」， **没解决断线重连**。而真正会丢结果的恰恰是后者：服务端收到 subscribe 会回一份 当前状态，掉线期间错过的推送就是靠这次重放补回来的；不重新订阅，重连后只收得到 「将来」的事件，可结果已经是过去式了。改成订阅意图保留到显式 `unsubscribe()`， 每次 onConnected 都重发。 这套逻辑原来只有 SubmissionWebSocket 有，FlowchartWebSocket 是 send 失败打一行 日志了事 —— socket 一掉，那次评分结果就再也回不来，按钮一直转圈。提成公共基类 SubscribingWebSocket，两条通道共用。 流程图另加轮询兜底：提交后 5 秒 WS 还没出结果就每 3 秒拉一次，读 status 2/3 结算，3 分钟上限。判题那边一直有兜底，流程图这边没有，而 Redis pub/sub 是发完 不管的，worker 推的那一刻连接不在就永远丢了。 `useSubmissionMonitor` 里 `watch(wsStatus, ..., { immediate: true })` 的回调在 watch() **返回之前**就同步跑了，已经连着时 `unwatch` 还是 null，if 不成立， watcher 永远停不掉：每提交一次泄漏一个，往后每次重连它们都会把各自那个早就判完 的旧 submissionId 重新订阅一遍。整块删掉，直接 subscribe —— 基类已经管了时序。 WS handler 原来不校验 submissionId。学生同时开着几道题的页面时，每条连接都订在 同一个用户 topic 上，别的页面的评分结果会被当成自己的。 ## 会话 握手时校验过一次会话就再也不管了，这条连接却能挂几个小时：用户在别的标签页 登出、或者会话本身到期，旧 socket 照样收推送。加一条 60 秒一轮的巡检，用 Redis EXPIRE 一条命令同时完成「判断存在」和「续期」（续期是必要的：只开着页面 挂 WS 的人一次 HTTP 请求都不发，不该被算成不活跃踢下线）。Redis 抛错时整轮 放弃，绝不因为一次抖动把全班踢下线。 禁用只改数据库的 isDisabled 列、不动 Redis 里的会话，巡检永远发现不了。加 `session:revoked` 频道主动通知，**两种作用域不能混**： { token } 用户登出。只断这一张会话 —— 同一个人在别的设备上是另一张 会话，按 userId 广播会把他手机上的登录一起踢掉 { userId } 账号被禁用。所有设备都得断 先发一帧 force_logout 再隔 100ms 断开。只断不发的话前端只看到一次普通掉线， 会照常重连、页面上还显示着登录态。token 不进帧里 —— 那是 httpOnly cookie 的值， 推到 WS 上就等于交给了 JS，匹配全在服务端做。 前端在协议层拦截 force_logout（和 pong 一样，不下发给业务 handler），并主动 disconnect —— 否则会一路 401 重连到退避上限，正是这机制要消掉的浪费。表现刻意 和 utils/api.ts 里 account-disabled / login-required 两支保持一致：同一件事从 HTTP 和 WS 两条路进来，学生看到的不该有两个样子。 ## 开销与安全 `void handleMessage(...)` 是裸的，里面有两次 DB 查询和一个会抛的 schema.parse， 库抖一下就是一个 unhandled rejection（隔壁 bridgeSubmissionEvents 两处都接住了， 只有这里漏了）。 ping 提到用户查询之前。原来的顺序是「先查 user 再看消息类型」，每个客户端每 30 秒都要为一次心跳打一趟数据库。禁用用户不会因此漏网：推送路径上 bridge 会查， subscribe 这条真正读数据的路径下面照样查。 bridge 两条 per-user 通道都先看 `server.subscriberCount(topic)`，没人订阅就别 查库了 —— 判题高峰期绝大多数事件的目标用户此刻并不在线。 flowchart 评分失败原来把 error.message 原样推给学生、前端直接弹出来，AI provider 的地址和内部报错就这么进了浏览器。改成真实原因写服务端日志。 加每连接令牌桶（20 突发 + 每秒回填 2）。一条 subscribe 在服务端是一到两次数据库 查询，一个学生开着一条 socket 狂发就能压住库。正常流量离阈值几十倍远。 升级时校验 Origin。会话 cookie 是 SameSite=Lax、WS 握手不是导航，跨站页面本来就 带不上 cookie，所以这是防御纵深不是唯一防线。同源放行；本机开发（Vite 5173 → API 3000）自动放行，且只在两边都是本机时成立 —— 生产环境 url.hostname 是正式 域名，这条永远不触发；跨域部署走 ALLOWED_WS_ORIGINS。不发 Origin 的一律放行： 真正的攻击面是带着受害者 cookie 的浏览器页面，而浏览器一定会带 Origin。 顺带：useConfigWebSocket 的 handler 从 onMounted 挪到同步注册（调用方在 setup 阶段就 connect() 了），删掉每条消息打完整内容的 console.log 和死字段 ws.data.username。 ## 没动的 题目页上并没有两条 /ws/submissions —— Form.vue 里 SubmitFlowchart 和 SubmitCode 是 v-if/v-else，互斥。学生实际是 2 条连接：全站一条 /ws/config + 一条 /ws/submissions，正常，不必合并。 ## 验证 55 个用例，分六组打桩跑（假 WebSocket + 假计时器；会话/限流/吊销三组对着真 Redis）： 重连语义 7 断开后不再自我复活、卸载后计时器已取消、退避封顶、 online 立即重连、旧 onclose 不污染新连接 订阅重放 9 未就绪时补发、重连后重新订阅、unsubscribe 后不再重放 会话巡检 8 EXPIRE 三态、只断失效的、同 token 只查一次、 Redis 抖动时一个都不踢 Origin/限流 13 跨站与跨端口拒绝、生产不因 localhost 开后门、 突发额度、按时间回填 强制登出 13 登出只断同 token（别的设备不受牵连）、禁用断所有设备、 巡检先通知再断 前端登出 5 两支表现、收到后不再重连、两条通道只处理一次 前两组做了改动前/后对比，老代码该挂的都挂了 —— 「重连后自动重新订阅」正是这么 跑出来的，此前我以为 pendingSubmissionId 已经覆盖了这种情况。 apps/api 的 tsc 和 apps/web 的 vue-tsc 都干净，仓库既有测试照常通过。 **SubmitFlowchart.vue 的轮询兜底未经运行时验证** —— 在 SFC 内，没搭组件挂载 环境，只过了类型检查和人工核对。要验的话，停掉 worker 提交一次流程图，看 5 秒 后是否转入轮询、3 分钟后是否给出超时提示。 这批改动动了 WS 的行为面（限流会断连接、Origin 会拒绝、巡检会踢会话），上线前 建议手测：提交代码看判题、切流程图看评分、后台改配置看全站生效、开两个标签页 在一个里登出、禁用一个在线学生看另一端反应。 Co-Authored-By: Claude Opus 5 --- .env.example | 4 + apps/api/src/auth/session.ts | 22 ++ apps/api/src/config.ts | 8 + apps/api/src/events.ts | 37 +++ apps/api/src/flowchart/run.ts | 6 +- apps/api/src/index.ts | 14 +- apps/api/src/routes/admin/account.ts | 7 + apps/api/src/routes/auth.ts | 7 +- apps/api/src/websocket.ts | 236 +++++++++++++-- .../oj/problem/components/SubmitFlowchart.vue | 122 +++++++- .../composables/useSubmissionMonitor.ts | 36 +-- apps/web/src/shared/composables/websocket.ts | 286 +++++++++++++----- 12 files changed, 647 insertions(+), 138 deletions(-) diff --git a/.env.example b/.env.example index d5bc974..ec27bb7 100644 --- a/.env.example +++ b/.env.example @@ -8,5 +8,9 @@ JUDGE_SERVER_URL=http://localhost:8081 # 留空的话后端会随机生成一个并在启动日志里告警，判题机心跳会被 403 挡掉。 JUDGE_SERVER_TOKEN= JUDGE_CONCURRENCY=2 +# WebSocket 升级额外放行的来源，逗号分隔的完整 origin。 +# 同源本来就放行，本机开发（Vite 5173 → API 3000）也自动放行，一般不用配。 +# 只有前后端分处不同域名时才需要，例如 ALLOWED_WS_ORIGINS=https://oj.example.com +ALLOWED_WS_ORIGINS= OJ2_DEV_USERNAME=student OJ2_DEV_PASSWORD=student123 diff --git a/apps/api/src/auth/session.ts b/apps/api/src/auth/session.ts index 3d047dc..98c4720 100644 --- a/apps/api/src/auth/session.ts +++ b/apps/api/src/auth/session.ts @@ -58,10 +58,12 @@ export async function createSession( }) } +/** 返回被删掉的 token：调用方要拿它去广播会话吊销，好断掉同一浏览器里其他标签页的连接 */ export async function destroySession(c: Context) { const token = getCookie(c, config.sessionCookie) if (token) await redis.del(sessionKey(token)) deleteCookie(c, config.sessionCookie, { path: "/" }) + return token ?? null } function readCookie(request: Request, name: string) { @@ -141,6 +143,26 @@ export async function getRequestSessionUser(request: Request) { return (await getUserByToken(readCookie(request, config.sessionCookie))).user } +/** + * WebSocket 升级时把 token 一起存进连接，之后才能定期确认这个会话还有效 + * —— 握手时校验过一次，可这条连接能挂上好几个小时。 + */ +export function readRequestSessionToken(request: Request) { + return readCookie(request, config.sessionCookie) ?? "" +} + +/** + * 会话还在就续期并返回 true，已登出或已过期返回 false。 + * + * 用 EXPIRE 一条命令同时完成「判断存在」和「续期」，比 GET + EXPIRE 少一趟往返。 + * 续期这件事本身也是要的：HTTP 请求会走 getUserByToken 里的 redis.expire 续期， + * 而只开着页面挂 WebSocket 的人一次请求都不发，不该因此被算成不活跃踢下线。 + */ +export async function touchSession(token: string) { + if (!token) return false + return (await redis.expire(sessionKey(token), config.sessionTtlSeconds)) === 1 +} + async function getStoredSession(c: Context) { const token = getCookie(c, config.sessionCookie) if (!token) return null diff --git a/apps/api/src/config.ts b/apps/api/src/config.ts index 4e1d812..adbf84b 100644 --- a/apps/api/src/config.ts +++ b/apps/api/src/config.ts @@ -76,6 +76,14 @@ export const config = { // 一言数据集（hitokoto.cn 官方导出），和旧后端读同一份：容器里是 /data/hitokoto。 // 本机 dev 默认路径下没有这份数据，读不到就回落到内置的几条，不影响启动。 hitokotoDirectory: repoPath(process.env.HITOKOTO_DIRECTORY ?? "data/hitokoto"), + /** + * WebSocket 升级时额外放行的来源（逗号分隔的完整 origin，如 https://oj.example.com）。 + * 同源本来就放行，只有前后端分处不同域名时才需要配。 + */ + allowedWebSocketOrigins: (process.env.ALLOWED_WS_ORIGINS ?? "") + .split(",") + .map((value) => value.trim()) + .filter(Boolean), uploadUriPrefix: process.env.UPLOAD_URI_PREFIX ?? "/public/upload", avatarUriPrefix: process.env.AVATAR_URI_PREFIX ?? "/public/avatar", aiBaseUrl: process.env.AI_BASE_URL ?? "https://api.deepseek.com", diff --git a/apps/api/src/events.ts b/apps/api/src/events.ts index 1df5e5f..3db2d5a 100644 --- a/apps/api/src/events.ts +++ b/apps/api/src/events.ts @@ -16,6 +16,43 @@ export async function publishConfigUpdate(key: string, value: unknown) { await redis.publish(configUpdateChannel, JSON.stringify({ type: "config_update", key, value })) } +/** + * 会话吊销广播。让还挂着的 WebSocket 立刻知道自己该下线了。 + * + * 两种作用域，**不能混**： + * - `{ token }` 用户登出。只该断这一张会话 —— 同一个人在别的设备上是另一张会话， + * 不该被牵连。 + * - `{ userId }` 账号被禁用。所有设备都得断。禁用只改数据库的 isDisabled 列、 + * 不动 Redis 里的会话，WebSocket 那边的会话巡检永远发现不了，只能靠这条主动通知。 + */ +export const sessionRevokedChannel = "session:revoked" + +export type SessionRevokedReason = "session-ended" | "account-disabled" + +export interface SessionRevoked { + token?: string + userId?: number + reason: SessionRevokedReason +} + +export async function publishSessionRevoked( + target: { token: string } | { userId: number }, + reason: SessionRevokedReason, +) { + await redis.publish(sessionRevokedChannel, JSON.stringify({ ...target, reason })) +} + +export function parseSessionRevoked(raw: string): SessionRevoked | null { + try { + const value = JSON.parse(raw) as SessionRevoked + if (typeof value.token !== "string" && !Number.isInteger(value.userId)) return null + if (value.reason !== "session-ended" && value.reason !== "account-disabled") return null + return value + } catch { + return null + } +} + interface UserEvent { userId: number data: FlowchartUpdate | Record diff --git a/apps/api/src/flowchart/run.ts b/apps/api/src/flowchart/run.ts index 1fef1d7..75b6182 100644 --- a/apps/api/src/flowchart/run.ts +++ b/apps/api/src/flowchart/run.ts @@ -65,12 +65,14 @@ export async function evaluateFlowchart(job: FlowchartJobData) { criteriaDetails: result.criteria, })) } catch (error) { - const message = error instanceof Error ? error.message : String(error) + // 原来这里把 error.message 原样推给学生、前端还直接 message.error 弹出来 —— + // AI provider 的地址、内部报错就这么进了浏览器。真实原因留在服务端日志里， + // 学生只需要知道「失败了，再试一次」；error 字段留空，前端有兜底文案。 + console.error(`Failed to evaluate flowchart ${row.flowchart.id}`, error) await db.update(schema.flowchartSubmission).set({ status: 3 }).where(eq(schema.flowchartSubmission.id, row.flowchart.id)) await publishFlowchartUpdate(row.flowchart.userId, flowchartUpdateSchema.parse({ type: "flowchart_evaluation_failed", submissionId: row.flowchart.id, - error: message, })) throw error } diff --git a/apps/api/src/index.ts b/apps/api/src/index.ts index 920c9dc..db78c5e 100644 --- a/apps/api/src/index.ts +++ b/apps/api/src/index.ts @@ -1,7 +1,7 @@ import { Hono } from "hono" import { basename, resolve } from "node:path" -import { getRequestSessionUser } from "./auth/session" +import { getRequestSessionUser, readRequestSessionToken } from "./auth/session" import { config } from "./config" import { adminRoutes } from "./routes/admin" import { authRoutes } from "./routes/auth" @@ -19,6 +19,8 @@ import { submissionRoutes } from "./routes/submission" import { siteRoutes } from "./routes/site" import { bridgeSubmissionEvents, + isAllowedWebSocketOrigin, + startSessionSweep, submissionWebSocketHandler, type SubmissionSocketData, } from "./websocket" @@ -99,12 +101,19 @@ const server = Bun.serve({ ) } if (url.pathname === "/ws/submissions" || url.pathname === "/ws/config") { + if (!isAllowedWebSocketOrigin(request.headers.get("origin"), url)) { + return new Response("Forbidden", { status: 403 }) + } const user = await getRequestSessionUser(request) if (!user) return new Response("Unauthorized", { status: 401 }) const kind = url.pathname === "/ws/config" ? "config" : "submissions" if ( bunServer.upgrade(request, { - data: { userId: user.id, username: user.username, kind }, + data: { + userId: user.id, + kind, + token: readRequestSessionToken(request), + }, }) ) { return undefined @@ -117,4 +126,5 @@ const server = Bun.serve({ }) await bridgeSubmissionEvents(server) +startSessionSweep() console.log(`OJ2 API listening on http://localhost:${server.port}`) diff --git a/apps/api/src/routes/admin/account.ts b/apps/api/src/routes/admin/account.ts index f6af4e0..0678027 100644 --- a/apps/api/src/routes/admin/account.ts +++ b/apps/api/src/routes/admin/account.ts @@ -17,6 +17,7 @@ import { requireSuperAdmin, type AppEnv } from "../../auth/middleware" import { db, schema } from "../../db" import { failure, success } from "../../http" import { queryInteger, sampleUser } from "../helpers" +import { publishSessionRevoked } from "../../events" export const adminAccountRoutes = new Hono() @@ -221,6 +222,12 @@ adminAccountRoutes.put("/users/:id", requireSuperAdmin, async (c) => { .where(eq(schema.userProfile.userId, id)) }) + // 禁用只改数据库这一列，不动 Redis 里的会话 —— 那个学生挂着的 WebSocket + // 靠会话巡检永远发现不了（token 还是好的），只能在这里主动断 + if (data.isDisabled && !existing.user.isDisabled) { + await publishSessionRevoked({ userId: id }, "account-disabled") + } + const [row] = await selectUser(id) return success(c, serialize(row!)) }) diff --git a/apps/api/src/routes/auth.ts b/apps/api/src/routes/auth.ts index 1723c79..afd3f51 100644 --- a/apps/api/src/routes/auth.ts +++ b/apps/api/src/routes/auth.ts @@ -4,6 +4,7 @@ import { Hono } from "hono" import { optionalAuth, type AppEnv } from "../auth/middleware" import { createSession, destroySession } from "../auth/session" +import { publishSessionRevoked } from "../events" import { hashPassword, verifyPassword } from "../auth/password" import { db, schema } from "../db" import { failure, success } from "../http" @@ -66,7 +67,11 @@ authRoutes.post("/auth/login", async (c) => { }) authRoutes.delete("/auth/session", async (c) => { - await destroySession(c) + const token = await destroySession(c) + // 同一个浏览器的其他标签页还挂着 WebSocket，页面上仍显示着登录态。推一条让它们 + // 立刻清掉，不用等最多 60 秒的会话巡检。 + // 按 token 而不是按用户：这个人在别的设备上的登录是另一张会话，不该被牵连。 + if (token) await publishSessionRevoked({ token }, "session-ended") return success(c, null) }) diff --git a/apps/api/src/websocket.ts b/apps/api/src/websocket.ts index 01508c9..0c1768e 100644 --- a/apps/api/src/websocket.ts +++ b/apps/api/src/websocket.ts @@ -1,6 +1,8 @@ import { flowchartUpdateSchema, submissionUpdateSchema } from "@oj2/contract" import { and, eq } from "drizzle-orm" +import { touchSession } from "./auth/session" +import { config } from "./config" import { db, schema } from "./db" import { parseSubmissionEvent, @@ -9,13 +11,147 @@ import { } from "./judge/events" import { JudgeStatus } from "./judge/status" import { createSubscriberRedis } from "./redis" -import { configTopic, configUpdateChannel, parseUserEvent, userEventChannel, userEventTopic } from "./events" +import { + configTopic, + configUpdateChannel, + parseSessionRevoked, + parseUserEvent, + sessionRevokedChannel, + userEventChannel, + userEventTopic, +} from "./events" + +/** 本机的几种写法。开发时 Vite 代理会让 Origin（5173）和 Host（3000）对不上 */ +const LOCAL_HOSTNAMES = new Set(["localhost", "127.0.0.1", "::1", "[::1]"]) + +/** + * WebSocket 升级的来源校验。 + * + * 会话 cookie 是 SameSite=Lax，而 WebSocket 握手不是导航，跨站页面本来就带不上 + * 这个 cookie —— 所以这里是防御纵深，不是唯一防线。 + * + * 不发 Origin 的一律放行：真正的攻击面是「带着受害者 cookie 的浏览器页面」， + * 而浏览器一定会带 Origin；脚本客户端本来就能伪造任意请求头，拦它没有意义。 + */ +export function isAllowedWebSocketOrigin(origin: string | null, url: URL) { + if (!origin) return true + if (config.allowedWebSocketOrigins.includes(origin)) return true + let originUrl: URL + try { + originUrl = new URL(origin) + } catch { + return false + } + if (originUrl.host === url.host) return true + // 两边都是本机才放行。生产环境 url.hostname 是正式域名，这条永远不成立 + return ( + LOCAL_HOSTNAMES.has(originUrl.hostname) && LOCAL_HOSTNAMES.has(url.hostname) + ) +} export interface SubmissionSocketData { userId: number - username: string /** 同一个 Bun.serve 只能挂一个 websocket handler，用它区分两条通道 */ kind: "submissions" | "config" + /** 握手时那张会话的 token，留着定期确认它还没被登出 / 过期，见 sweepSessions */ + token: string + /** 令牌桶，open 时初始化，见 allowMessage */ + rate?: { tokens: number; updatedAt: number } +} + +/** + * 每条连接的消息限流。 + * + * 一条 subscribe 在服务端是一到两次数据库查询，一个学生开着一条 socket 狂发就能 + * 压住库。正常流量离这个阈值很远：心跳 30 秒一条，订阅一次提交也就一两条， + * 20 的突发额度 + 每秒 2 个的回填是几十倍的余量。 + */ +const RATE_BURST = 20 +const RATE_REFILL_PER_SECOND = 2 + +function allowMessage(ws: Bun.ServerWebSocket) { + const now = Date.now() + const rate = (ws.data.rate ??= { tokens: RATE_BURST, updatedAt: now }) + const refill = ((now - rate.updatedAt) / 1000) * RATE_REFILL_PER_SECOND + rate.tokens = Math.min(RATE_BURST, rate.tokens + refill) + rate.updatedAt = now + if (rate.tokens < 1) return false + rate.tokens -= 1 + return true +} + +/** + * 当前挂着的连接。Bun 不提供遍历连接的接口，要定期巡检就得自己登记。 + * open 时加入、close 时移除，见 sweepSessions。 + */ +const liveSockets = new Set>() + +/** 会话巡检间隔。够快到登出后一分钟内断开，又不至于让 Redis 忙起来 */ +const SESSION_SWEEP_INTERVAL = 60_000 + +/** 先把 force_logout 帧发出去，再断连接，留一拍给它出门 */ +const FORCE_LOGOUT_CLOSE_DELAY = 100 + +/** + * 通知并断开一批连接。 + * + * 之所以先发一帧再断：只断连接的话前端只看到一次普通掉线，会照常重连，页面上 + * 还显示着登录态；收到 force_logout 才知道要清掉身份、弹登录框或者提示被禁用。 + */ +function forceLogout( + targets: Bun.ServerWebSocket[], + reason: string, +) { + if (targets.length === 0) return + const frame = JSON.stringify({ type: "force_logout", reason }) + for (const ws of targets) ws.send(frame) + setTimeout(() => { + for (const ws of targets) ws.close(1008, "Session ended") + }, FORCE_LOGOUT_CLOSE_DELAY) +} + +/** + * 定期把会话已经失效的连接断掉。 + * + * 握手时校验过一次会话，但这条连接能挂几个小时 —— 期间用户可能在别的标签页登出， + * 或者会话本身到期。只靠消息触发的校验不够：一条连接完全可能除了心跳什么都不发， + * 而心跳是**故意**不查会话的（否则每客户端每 30 秒一趟 Redis 又回来了）。 + * + * 注意这里不查 isDisabled：管理员禁用只改数据库列、不删会话，所以 token 校验 + * 覆盖不到它。禁用由推送路径上的 bridgeSubmissionEvents 挡着 —— 被禁用的学生 + * 收不到任何数据，socket 还挂着只是根空管子。 + */ +export async function sweepSessions() { + // 一个学生至少有配置和提交两条通道，多开几个标签页还会更多，而它们共用同一张 + // 会话 —— 一轮里同一个 token 只查一次 + const checked = new Map() + const dead: Bun.ServerWebSocket[] = [] + for (const ws of liveSockets) { + const token = ws.data.token + let alive = checked.get(token) + if (alive === undefined) { + try { + alive = await touchSession(token) + } catch (error) { + // Redis 抖一下不该把全班踢下线：这一轮直接放弃，下一轮再说 + console.error("Failed to verify websocket sessions", error) + return + } + checked.set(token, alive) + } + if (!alive) dead.push(ws) + } + // 会话没了有两种可能：在别的标签页登出了，或者会话自己到期。对用户都是 + // 「要重新登录」，走 session-ended 这一支 + forceLogout(dead, "session-ended") +} + +export function startSessionSweep() { + const timer = setInterval(() => { + void sweepSessions() + }, SESSION_SWEEP_INTERVAL) + timer.unref() + return timer } function objectValue(value: unknown): Record { @@ -27,6 +163,8 @@ function objectValue(value: unknown): Record { export function submissionWebSocketHandler(): Bun.WebSocketHandler { return { open(ws) { + liveSockets.add(ws) + ws.data.rate = { tokens: RATE_BURST, updatedAt: Date.now() } if (ws.data.kind === "config") { ws.subscribe(configTopic) return @@ -35,9 +173,20 @@ export function submissionWebSocketHandler(): Bun.WebSocketHandler { + console.error("Failed to handle websocket message", error) + ws.send(JSON.stringify({ type: "error", message: "Internal error" })) + }) }, close(ws) { + liveSockets.delete(ws) if (ws.data.kind === "config") { ws.unsubscribe(configTopic) return @@ -52,6 +201,34 @@ async function handleMessage( ws: Bun.ServerWebSocket, raw: string, ) { + let message: { type?: unknown; timestamp?: unknown; submissionId?: unknown } + try { + message = JSON.parse(raw) as typeof message + } catch { + ws.send(JSON.stringify({ type: "error", message: "Invalid JSON" })) + return + } + + // 心跳不查库。原来的顺序是「先查 user 再看消息类型」，于是每个客户端每 30 秒 + // 都要为一次 ping 打一趟数据库；一个题目页还开着两条连接，全班在线时纯空转。 + // 禁用用户不会因此漏网：往用户 topic 推之前 bridgeSubmissionEvents 会查一次， + // 而 subscribe 这条真正读数据的路径下面照样查。 + if (message.type === "ping") { + ws.send(JSON.stringify({ type: "pong", timestamp: message.timestamp })) + return + } + if (message.type !== "subscribe" || typeof message.submissionId !== "string") { + ws.send(JSON.stringify({ type: "error", message: "Invalid message" })) + return + } + + // 会话可能在连接期间就失效了：用户在别的标签页登出，或者会话自己到期。 + // 握手时校验过一次不算数 —— 这条连接能挂几个小时。 + if (!(await touchSession(ws.data.token))) { + ws.close(1008, "Session expired") + return + } + const [activeUser] = await db .select({ id: schema.user.id }) .from(schema.user) @@ -67,23 +244,6 @@ async function handleMessage( return } - let message: { type?: unknown; timestamp?: unknown; submissionId?: unknown } - try { - message = JSON.parse(raw) as typeof message - } catch { - ws.send(JSON.stringify({ type: "error", message: "Invalid JSON" })) - return - } - - if (message.type === "ping") { - ws.send(JSON.stringify({ type: "pong", timestamp: message.timestamp })) - return - } - if (message.type !== "subscribe" || typeof message.submissionId !== "string") { - ws.send(JSON.stringify({ type: "error", message: "Invalid message" })) - return - } - const [submission] = await db .select({ id: schema.submission.id, @@ -112,7 +272,7 @@ async function handleMessage( const replay = flowchart.status === 2 ? { type: "flowchart_evaluation_completed", submissionId: flowchart.id, score: flowchart.score ?? undefined, grade: flowchart.grade ?? undefined } : flowchart.status === 3 - ? { type: "flowchart_evaluation_failed", submissionId: flowchart.id, error: "Evaluation failed" } + ? { type: "flowchart_evaluation_failed", submissionId: flowchart.id } : { type: "flowchart_evaluation_update", submissionId: flowchart.id } ws.send(JSON.stringify(flowchartUpdateSchema.parse(replay))) return @@ -147,9 +307,27 @@ export async function bridgeSubmissionEvents( server.publish(configTopic, raw) return } + if (channel === sessionRevokedChannel) { + const revoked = parseSessionRevoked(raw) + if (!revoked) return + // 按 token 还是按 userId，取决于是「这张会话登出了」还是「这个账号被禁用了」 + forceLogout( + [...liveSockets].filter((ws) => + revoked.token !== undefined + ? ws.data.token === revoked.token + : ws.data.userId === revoked.userId, + ), + revoked.reason, + ) + return + } if (channel === userEventChannel) { const event = parseUserEvent(raw) if (!event) return + const topic = userEventTopic(event.userId) + // 这台实例上没人订阅就到此为止：判题高峰期绝大多数事件的目标用户此刻并不 + // 在线，查一次库只为了 publish 给零个订阅者 + if (server.subscriberCount(topic) === 0) return void (async () => { const [activeUser] = await db .select({ id: schema.user.id }) @@ -157,7 +335,7 @@ export async function bridgeSubmissionEvents( .where(and(eq(schema.user.id, event.userId), eq(schema.user.isDisabled, false))) .limit(1) if (!activeUser) return - server.publish(userEventTopic(event.userId), JSON.stringify(event.data)) + server.publish(topic, JSON.stringify(event.data)) })().catch((error) => { console.error("Failed to bridge user event", error) }) @@ -166,6 +344,8 @@ export async function bridgeSubmissionEvents( if (channel !== submissionUpdateChannel) return const event = parseSubmissionEvent(raw) if (!event) return + const topic = userSubmissionTopic(event.userId) + if (server.subscriberCount(topic) === 0) return void (async () => { const [activeUser] = await db .select({ id: schema.user.id }) @@ -178,10 +358,7 @@ export async function bridgeSubmissionEvents( ) .limit(1) if (!activeUser) return - server.publish( - userSubmissionTopic(event.userId), - JSON.stringify(event.data), - ) + server.publish(topic, JSON.stringify(event.data)) })().catch((error) => { console.error("Failed to bridge submission event", error) }) @@ -189,6 +366,11 @@ export async function bridgeSubmissionEvents( subscriber.on("error", (error) => { console.error("Submission event subscriber error", error) }) - await subscriber.subscribe(submissionUpdateChannel, userEventChannel, configUpdateChannel) + await subscriber.subscribe( + submissionUpdateChannel, + userEventChannel, + configUpdateChannel, + sessionRevokedChannel, + ) return subscriber } diff --git a/apps/web/src/oj/problem/components/SubmitFlowchart.vue b/apps/web/src/oj/problem/components/SubmitFlowchart.vue index b6f3e41..a816897 100644 --- a/apps/web/src/oj/problem/components/SubmitFlowchart.vue +++ b/apps/web/src/oj/problem/components/SubmitFlowchart.vue @@ -17,6 +17,7 @@ import { useMyFlowchartStore } from "shared/store/myFlowchart" // API 和状态管理 import { getCurrentProblemFlowchartSubmission, + getFlowchartSubmission, getFlowchartSubmissionDetail, submitFlowchart, } from "oj/api" @@ -87,30 +88,129 @@ function splitSuggestionLines(suggestions?: string | null) { : [] } +// ==================== 评分结果监听 ==================== +/** + * 评分结果有两条路进来：WebSocket 推送（快）和轮询（稳）。谁先到谁结算， + * 靠 monitoringId 去重 —— 结算时清空，另一条路后到就直接跳过。 + * + * 之所以必须有轮询兜底：Redis pub/sub 是发完不管的，worker 推的那一刻只要这条 + * 连接不在（重连空窗、页面刚从后台切回来），这条消息就永远丢了。判题那边一直 + * 有兜底轮询，流程图这边没有，丢一次消息按钮就一直转圈转到用户自己刷新。 + */ +const monitoringId = ref("") + +/** AI 评分比判题慢得多，轮询放缓一点 */ +const POLL_INTERVAL = 3000 +/** 到点还没结果就收手，别无限轮询下去 */ +const POLL_TIMEOUT = 3 * 60 * 1000 + +type Outcome = + | { ok: true; score: number; grade: string } + | { ok: false; error?: string } + +const { pause: pausePolling, resume: resumePolling } = useIntervalFn( + async () => { + if (!monitoringId.value) { + pausePolling() + return + } + try { + const data = await getFlowchartSubmission(monitoringId.value) + if (data.status === 2) { + settle(data.id, { + ok: true, + score: data.aiScore ?? 0, + grade: data.aiGrade ?? "", + }) + } else if (data.status === 3) { + settle(data.id, { ok: false }) + } + } catch (error) { + console.error("[Flowchart] 轮询失败:", error) + pausePolling() + } + }, + POLL_INTERVAL, + { immediate: false }, +) + +// WebSocket 正常时压根用不上轮询，先给它 5 秒，到点还没结果才开始拉 +const { start: startPollingFallback, stop: stopPollingFallback } = useTimeoutFn( + () => { + if (monitoringId.value) resumePolling() + }, + 5000, + { immediate: false }, +) + +const { start: startPollingDeadline, stop: stopPollingDeadline } = useTimeoutFn( + () => { + if (!monitoringId.value) return + monitoringId.value = "" + unsubscribe() + pausePolling() + loading.value = false + message.warning("评分等待超时，请稍后刷新页面查看结果") + }, + POLL_TIMEOUT, + { immediate: false }, +) + +function settle(submissionId: string, outcome: Outcome) { + // 一个学生可能同时开着几道题的页面，每条连接都订在同一个用户 topic 上， + // 别的页面的评分结果照样会推到这里来 —— 必须认 id，不然会张冠李戴 + if (!submissionId || submissionId !== monitoringId.value) return + monitoringId.value = "" + unsubscribe() + pausePolling() + stopPollingFallback() + stopPollingDeadline() + loading.value = false + + if (!outcome.ok) { + message.error( + outcome.error + ? `流程图评分失败: ${outcome.error}` + : "流程图评分失败，请稍后重试", + ) + return + } + latestRating.value = { score: outcome.score, grade: outcome.grade } + message.success( + `流程图评分完成！得分: ${outcome.score}分 (${outcome.grade}级)`, + ) + if ( + (outcome.grade === "A" || outcome.grade === "S") && + lastSubmittedMermaidCode.value + ) { + myFlowchartStore.show(lastSubmittedMermaidCode.value) + } +} + // ==================== WebSocket 相关函数 ==================== const handleWebSocketMessage = (data: FlowchartEvaluationUpdate) => { if (data.type === "flowchart_evaluation_completed") { - loading.value = false - const grade = data.grade || "" - latestRating.value = { score: data.score || 0, grade } - message.success(`流程图评分完成！得分: ${data.score}分 (${grade}级)`) - if ((grade === "A" || grade === "S") && lastSubmittedMermaidCode.value) { - myFlowchartStore.show(lastSubmittedMermaidCode.value) - } + settle(data.submissionId, { + ok: true, + score: data.score ?? 0, + grade: data.grade || "", + }) } else if (data.type === "flowchart_evaluation_failed") { - loading.value = false - message.error(`流程图评分失败: ${data.error}`) + settle(data.submissionId, { ok: false, error: data.error }) } } // 创建 WebSocket 连接 -const { connect, disconnect, subscribe } = useFlowchartWebSocket( +const { connect, disconnect, subscribe, unsubscribe } = useFlowchartWebSocket( handleWebSocketMessage, ) -// 订阅提交更新 +// 订阅提交更新，同时开启轮询兜底 function subscribeToSubmission(submissionId: string) { + monitoringId.value = submissionId subscribe(submissionId) + startPollingFallback() + startPollingDeadline() } // ==================== 提交相关函数 ==================== diff --git a/apps/web/src/oj/problem/composables/useSubmissionMonitor.ts b/apps/web/src/oj/problem/composables/useSubmissionMonitor.ts index eb675bc..68e2f18 100644 --- a/apps/web/src/oj/problem/composables/useSubmissionMonitor.ts +++ b/apps/web/src/oj/problem/composables/useSubmissionMonitor.ts @@ -1,4 +1,4 @@ -import { ref, computed, watch, onUnmounted } from "vue" +import { ref, computed, onUnmounted } from "vue" import { useIntervalFn, useTimeoutFn } from "@vueuse/core" import { getSubmission } from "oj/api" import { SubmissionStatus } from "utils/constants" @@ -81,6 +81,8 @@ export function useSubmissionMonitor() { // 停止轮询（WebSocket已成功） pausePolling() + // 结果已经到手，别让重连再去重放这条早就判完的订阅 + unsubscribe() getSubmission(submissionId.value).then((res) => { submission.value = res @@ -94,9 +96,9 @@ export function useSubmissionMonitor() { const { connect, subscribe, + unsubscribe, scheduleDisconnect, cancelScheduledDisconnect, - status: wsStatus, } = useSubmissionWebSocket(handleSubmissionUpdate) // ==================== 轮询保底启动 ==================== @@ -124,27 +126,17 @@ export function useSubmissionMonitor() { // 取消之前的断开计划 cancelScheduledDisconnect() - // 如果WebSocket未连接，先连接 - if (wsStatus.value !== "connected") { - console.log("[SubmissionMonitor] 启动WebSocket连接...") - connect() - } + // connect() 是幂等的：已经连着就直接返回，顺带把上一次空闲断开留下的状态清掉 + connect() - // 等待WebSocket连接并订阅 - let unwatch: (() => void) | null = null - unwatch = watch( - wsStatus, - (status) => { - if (status === "connected") { - console.log("[SubmissionMonitor] WebSocket已连接，订阅提交:", id) - subscribe(id) - if (unwatch) { - unwatch() // 订阅成功后停止监听 - } - } - }, - { immediate: true }, - ) + // 直接订阅，不必先等 status 变成 connected：连接没就绪时 subscribe() 会把 id + // 记在 pendingSubmissionId 上，由 onConnected() 补发（断线重连后同样有效）。 + // + // 原来这里是 watch(wsStatus, ..., { immediate: true })，而 immediate 的回调在 + // watch() **返回之前**就同步跑了 —— 已经连着时 unwatch 还是 null，if 不成立， + // 这个 watcher 就永远停不掉：每提交一次泄漏一个，而且往后每次重连时它们都会 + // 把各自那个早就判完的旧 submissionId 重新订阅一遍。 + subscribe(id) // 5秒后启动轮询保底（防止WebSocket失败） startPollingFallback() diff --git a/apps/web/src/shared/composables/websocket.ts b/apps/web/src/shared/composables/websocket.ts index 98c1340..a3f9478 100644 --- a/apps/web/src/shared/composables/websocket.ts +++ b/apps/web/src/shared/composables/websocket.ts @@ -1,5 +1,37 @@ +import { createDiscreteApi } from "naive-ui" import { ref, onUnmounted, type Ref } from "vue" +import { useAuthModalStore } from "shared/store/authModal" +import { useUserStore } from "shared/store/user" +import { STORAGE_KEY } from "utils/constants" +import storage from "utils/storage" + +// 全站唯一一处，脱离 n-message-provider 也能弹 —— 强制登出跟当前挂着哪个组件无关。 +// utils/api.ts 里也是这么做的 +const { message: toast } = createDiscreteApi(["message"]) + +/** + * 服务端要求下线。两种来源：账号被管理员禁用，或者这张会话没了（在别的标签页 + * 登出、或者会话到期）。 + * + * 表现刻意和 utils/api.ts 里 account-disabled / login-required 两支保持一致 —— + * 同一件事从 HTTP 和 WebSocket 两条路进来，学生看到的结果不该有两个样子。 + */ +function handleForceLogout(reason: string) { + const userStore = useUserStore() + // 配置通道和提交通道可能同时挂着，两条都会收到这一帧。第一次就把登录态清了， + // 第二次在这里掉头，免得弹两遍 + if (!userStore.isAuthed) return + storage.remove(STORAGE_KEY.AUTHED) + userStore.clearProfile() + if (reason === "account-disabled") { + // 不能弹登录框：账号已经禁用，登进去还是被拒，会陷进「弹框 → 登录 → 又弹框」 + toast.error("账号已被禁用，请联系老师") + return + } + useAuthModalStore().openLoginModal() +} + /** * WebSocket 连接状态 */ @@ -20,10 +52,16 @@ export interface WebSocketMessage { export interface WebSocketConfig { /** 完整 URL。后端只认 /ws/submissions 和 /ws/config 两条，按当前页面的协议与 host 拼 */ url: string - /** 最大重连次数，默认 5 */ + /** + * 最大重连次数，默认不限。 + * 原来默认 5 次、线性退避，加起来只有 15 秒 —— 后端 deploy 重启一次就超了， + * 之后这条连接死到用户刷新页面为止。机房网络抖动同理，所以默认不再封顶。 + */ maxReconnectAttempts?: number - /** 重连延迟（毫秒），默认 1000 */ + /** 首次重连延迟（毫秒），默认 1000。之后指数退避 */ reconnectDelay?: number + /** 重连延迟上限（毫秒），默认 30000 */ + maxReconnectDelay?: number /** 心跳间隔（毫秒），默认 30000（30秒） */ heartbeatTime?: number /** 是否启用心跳，默认 true */ @@ -54,15 +92,25 @@ export class BaseWebSocket { protected heartbeatTime: number protected enableHeartbeat: boolean protected enableAutoReconnect: boolean + protected maxReconnectDelay: number protected disconnectTimer: number | null = null + protected reconnectTimer: number | null = null + /** + * 「用户主动断开」的意图，和 enableAutoReconnect 这个**配置**分开存。 + * 以前两者共用一个字段：disconnect() 把配置改成 false 来阻止重连，而 connect() + * 从不改回 true —— 登出再登录后，这条连接就永远失去了自动重连能力。 + */ + protected closedByUser = false + protected reviveBound = false public status: Ref = ref("disconnected") constructor(config: WebSocketConfig) { this.url = config.url - this.maxReconnectAttempts = config.maxReconnectAttempts ?? 5 + this.maxReconnectAttempts = config.maxReconnectAttempts ?? Number.POSITIVE_INFINITY this.reconnectDelay = config.reconnectDelay ?? 1000 + this.maxReconnectDelay = config.maxReconnectDelay ?? 30000 this.heartbeatTime = config.heartbeatTime ?? 30000 this.enableHeartbeat = config.enableHeartbeat ?? true this.enableAutoReconnect = config.enableAutoReconnect ?? true @@ -72,6 +120,10 @@ export class BaseWebSocket { * 连接 WebSocket */ connect() { + // 重新表达「我要连着」的意图：把上一次 disconnect() 留下的状态清掉 + this.closedByUser = false + this.clearReconnectTimer() + if ( this.ws && (this.ws.readyState === WebSocket.OPEN || @@ -80,12 +132,17 @@ export class BaseWebSocket { return } + this.bindReviveListeners() this.status.value = "connecting" try { - this.ws = new WebSocket(this.url) + // 所有回调都闭包住这个局部 ws 而不是读 this.ws：一条被换掉的旧连接 + // 迟到的 onclose / onerror 不该去改现在这条连接的状态 + const ws = new WebSocket(this.url) + this.ws = ws - this.ws.onopen = () => { + ws.onopen = () => { + if (ws !== this.ws) return this.status.value = "connected" this.reconnectAttempts = 0 console.log(`[WebSocket] 连接成功: ${this.url}`) @@ -95,16 +152,26 @@ export class BaseWebSocket { this.onConnected() } - this.ws.onmessage = (event) => { + ws.onmessage = (event) => { + if (ws !== this.ws) return try { const data = JSON.parse(event.data) as T - console.log(`[WebSocket] 收到消息:`, data) // 处理心跳响应 if (data.type === "pong") { return } + // 服务端要求下线。和 pong 一样是协议层的事，不该让每个业务 handler + // 各自认一遍 —— 而且此刻多半根本没有能处理它的 handler 挂着 + if (data.type === "force_logout") { + // 必须主动断，否则服务端断开后这条连接会照常自动重连，然后一路 401 + // 撞到退避上限 —— 正是这条机制要消掉的浪费 + this.disconnect() + handleForceLogout(String(data.reason ?? "")) + return + } + // 调用消息处理钩子 this.onMessage(data) } catch (error) { @@ -112,50 +179,112 @@ export class BaseWebSocket { } } - this.ws.onerror = (error) => { + ws.onerror = (error) => { + if (ws !== this.ws) return console.error("[WebSocket] 连接错误:", error) this.status.value = "error" this.onError(error) } - this.ws.onclose = (event) => { + ws.onclose = (event) => { + // disconnect() 会先把 this.ws 置空再 close()，所以主动断开走不到这里， + // 收尾由 disconnect() 自己做 —— 也就不会再像以前那样「断完立刻重连」 + if (ws !== this.ws) return + this.ws = null console.log( `[WebSocket] 连接关闭: code=${event.code}, reason=${event.reason}`, ) this.status.value = "disconnected" this.stopHeartbeat() this.onDisconnected(event) - - // 自动重连 - if ( - this.enableAutoReconnect && - this.reconnectAttempts < this.maxReconnectAttempts - ) { - this.reconnectAttempts++ - const delay = this.reconnectDelay * this.reconnectAttempts - console.log( - `[WebSocket] 将在 ${delay}ms 后重连 (尝试 ${this.reconnectAttempts}/${this.maxReconnectAttempts})`, - ) - setTimeout(() => this.connect(), delay) - } + this.scheduleReconnect() } } catch (error) { console.error("Failed to create WebSocket connection:", error) this.status.value = "error" + this.scheduleReconnect() } } + /** + * 安排一次重连。指数退避 + 抖动：一个班几十台机器同时掉线时， + * 别在同一毫秒一起冲回来把刚起来的后端再压趴一次。 + */ + protected scheduleReconnect() { + if (this.closedByUser || !this.enableAutoReconnect) return + if (this.reconnectAttempts >= this.maxReconnectAttempts) return + if (this.reconnectTimer !== null) return + + this.reconnectAttempts++ + const base = Math.min( + this.reconnectDelay * 2 ** (this.reconnectAttempts - 1), + this.maxReconnectDelay, + ) + const delay = Math.round(base * (0.5 + Math.random() * 0.5)) + console.log(`[WebSocket] 将在 ${delay}ms 后重连 (第 ${this.reconnectAttempts} 次)`) + this.reconnectTimer = window.setTimeout(() => { + this.reconnectTimer = null + this.connect() + }, delay) + } + + protected clearReconnectTimer() { + if (this.reconnectTimer !== null) { + clearTimeout(this.reconnectTimer) + this.reconnectTimer = null + } + } + + /** + * 网络恢复 / 标签页重新可见时立刻重连，不必等退避计时器走完。 + * 退避到 30 秒后，用户切回页面却还要再干等半分钟是说不过去的。 + */ + protected readonly revive = () => { + if (this.closedByUser || !this.enableAutoReconnect) return + if ( + this.ws && + (this.ws.readyState === WebSocket.OPEN || + this.ws.readyState === WebSocket.CONNECTING) + ) { + return + } + if (document.visibilityState === "hidden") return + if (navigator.onLine === false) return + this.clearReconnectTimer() + this.reconnectAttempts = 0 + this.connect() + } + + protected bindReviveListeners() { + if (this.reviveBound) return + this.reviveBound = true + window.addEventListener("online", this.revive) + document.addEventListener("visibilitychange", this.revive) + } + + protected unbindReviveListeners() { + if (!this.reviveBound) return + this.reviveBound = false + window.removeEventListener("online", this.revive) + document.removeEventListener("visibilitychange", this.revive) + } + /** * 断开连接 */ disconnect() { + this.closedByUser = true this.cancelScheduledDisconnect() + // 以前没存重连计时器的句柄：卸载后那个 setTimeout 照样会触发 connect()， + // 在已经销毁的组件上又建一条连接出来 + this.clearReconnectTimer() this.stopHeartbeat() - this.enableAutoReconnect = false // 停止自动重连 - if (this.ws) { - this.ws.close() - this.ws = null - } + this.unbindReviveListeners() + this.reconnectAttempts = 0 + // 先摘掉引用再 close()，onclose 里的 `ws !== this.ws` 就能识别出这是主动断开 + const ws = this.ws + this.ws = null + if (ws) ws.close() this.status.value = "disconnected" } @@ -169,11 +298,14 @@ export class BaseWebSocket { // 设置新的定时器 this.disconnectTimer = window.setTimeout(() => { + this.disconnectTimer = null const minutes = Math.floor(delay / 60000) console.log(`WebSocket idle for ${minutes} minutes, disconnecting...`) + // 这里**只断开**。原来断完紧接着一句 `enableAutoReconnect = true`， + // 而 close 是异步的 —— 等 onclose 跑到时标志已经翻回来了，于是 1 秒后 + // 又自动连上：这个「省资源」的空闲断开从来没有真正生效过。 + // 下一次 connect()（新提交）会自己把 closedByUser 清掉，不需要在这里预置。 this.disconnect() - // 断开后需要重新允许自动重连 - this.enableAutoReconnect = true }, delay) } @@ -294,39 +426,57 @@ export interface SubmissionUpdate extends WebSocketMessage { } /** - * 提交 WebSocket 连接管理类 + * 带「订阅意图」的连接。 + * + * subscribe() 在连接还没就绪时先把 id 记下来，等 onConnected() 补发 —— 调用方 + * 不用关心此刻连上没有，断线重连后也会自动重新订阅。 + * + * 原来这套只有 SubmissionWebSocket 有，FlowchartWebSocket 是 send 失败就打一行 + * 日志了事：socket 一掉，那次评分的结果就再也回不来，页面永远转圈。提到基类上， + * 两条通道共用同一套语义。 */ -class SubmissionWebSocket extends BaseWebSocket { - private pendingSubmissionId = "" - - constructor() { - const protocol = window.location.protocol === "https:" ? "wss:" : "ws:" - super({ url: `${protocol}//${window.location.host}/ws/submissions` }) - } +class SubscribingWebSocket< + T extends WebSocketMessage, +> extends BaseWebSocket { + /** + * 当前在等结果的提交。**一直留着**，直到调用方 unsubscribe()。 + * + * 原来这个字段叫 pendingSubmissionId，订阅一发成功就清空 —— 它只解决了 + * 「还没连上就调 subscribe」，没解决断线重连。而真正会丢结果的恰恰是后者： + * 服务端收到 subscribe 会回一份当前状态，掉线期间错过的那条推送就是靠这次 + * 重放补回来的。不重新订阅，重连后就只收得到「将来」的事件，可结果已经是过去式了。 + */ + private subscribedId = "" /** - * 订阅特定提交的更新 + * 订阅特定提交的更新。连接没就绪也可以调，连上后会自动补发。 */ subscribe(submissionId: string) { - this.pendingSubmissionId = submissionId - const success = this.send({ - type: "subscribe", - submissionId, - }) - if (success) this.pendingSubmissionId = "" + this.subscribedId = submissionId + this.sendSubscribe(submissionId) + } + + /** 结果已经拿到，重连后不必再问一遍 */ + unsubscribe() { + this.subscribedId = "" } protected onConnected() { - if (!this.pendingSubmissionId) return - const submissionId = this.pendingSubmissionId - if ( - this.send({ - type: "subscribe", - submissionId, - }) - ) { - this.pendingSubmissionId = "" - } + if (this.subscribedId) this.sendSubscribe(this.subscribedId) + } + + private sendSubscribe(submissionId: string) { + return this.send({ type: "subscribe", submissionId }) + } +} + +/** + * 提交 WebSocket 连接管理类 + */ +class SubmissionWebSocket extends SubscribingWebSocket { + constructor() { + const protocol = window.location.protocol === "https:" ? "wss:" : "ws:" + super({ url: `${protocol}//${window.location.host}/ws/submissions` }) } } @@ -356,6 +506,7 @@ export function useSubmissionWebSocket( connect: () => ws.connect(), disconnect: () => ws.disconnect(), subscribe: (submissionId: string) => ws.subscribe(submissionId), + unsubscribe: () => ws.unsubscribe(), scheduleDisconnect: (delay?: number) => ws.scheduleDisconnect(delay), cancelScheduledDisconnect: () => ws.cancelScheduledDisconnect(), status: ws.status, @@ -438,24 +589,11 @@ export interface FlowchartEvaluationUpdate extends WebSocketMessage { /** * 流程图 WebSocket 连接管理类 */ -class FlowchartWebSocket extends BaseWebSocket { +class FlowchartWebSocket extends SubscribingWebSocket { constructor() { const protocol = window.location.protocol === "https:" ? "wss:" : "ws:" super({ url: `${protocol}//${window.location.host}/ws/submissions` }) } - - /** - * 订阅特定流程图提交的更新 - */ - subscribe(submissionId: string) { - const success = this.send({ - type: "subscribe", - submissionId, - }) - if (!success) { - console.error("[Flowchart WebSocket] 订阅失败: 连接未就绪") - } - } } /** @@ -483,6 +621,7 @@ export function useFlowchartWebSocket( connect: () => ws.connect(), disconnect: () => ws.disconnect(), subscribe: (submissionId: string) => ws.subscribe(submissionId), + unsubscribe: () => ws.unsubscribe(), scheduleDisconnect: (delay?: number) => ws.scheduleDisconnect(delay), cancelScheduledDisconnect: () => ws.cancelScheduledDisconnect(), status: ws.status, @@ -521,11 +660,12 @@ class ConfigWebSocket extends BaseWebSocket { export function useConfigWebSocket(handler?: MessageHandler) { const ws = new ConfigWebSocket() - onMounted(() => { - if (handler) { - ws.addHandler(handler) - } - }) + // 同步注册，和另外两个 composable 一致。原来放在 onMounted 里，而调用方 + // （useConfigUpdate）在 setup 阶段就 connect() 了 —— 中间那段窗口收到的广播 + // 没有任何 handler 接。窗口极小，但没有任何理由留着它。 + if (handler) { + ws.addHandler(handler) + } onUnmounted(() => { if (handler) {