perf(redis): 热路径合并往返，限流脚本改走 EVALSHA

审查 Redis 用法时找到的四处小账，都是确定性改动，语义不变：

- `getUserByToken` 里两条串行 EXPIRE 合成一次 pipeline。这是全后端最热的
  Redis 路径 —— 每个带鉴权的 HTTP 请求都要续一次会话和反向索引。上次给
  `touchSession` 修的正是同一个形状，这处漏了，两边现在一致。
- `createSession` 的 SET / SADD / EXPIRE 三趟并成一趟。登录是突发的，
  一个班同时登录时差别全压在这一下。
- 限流的 Lua 从 `redis.eval` 改成 `defineCommand`，稳态走 EVALSHA 只发
  40 字节 sha1，不再每次带上 937 字节的脚本全文；NOSCRIPT 由 ioredis
  自动回退成 EVAL 重新灌，Redis 重启和 SCRIPT FLUSH 都不用管。
- 删掉 websocket 订阅连接上重复的 error 监听 —— `withErrorLogging` 已经
  打过一遍且带连接名，留着只会把同一条错误打两份。

实跑验证（dev 栈，API 跑在 3999）：登录后 `session:*` 多一条、
`user-sessions:1` 的 scard 和 ttl(604800) 都对；把两个键的 TTL 压到 100
再打一次带鉴权的请求，两个都回到 604800。限流侧 `info commandstats` 显示
evalsha calls=3/failed=1 + eval calls=2 —— 失败那次正是 SCRIPT FLUSH 之后
的 NOSCRIPT 回退；令牌桶数值与旧实现逐位一致（10 个初始额度扣 3 剩 7，
再要 20 个被拒并返回 wait=433.33 = (20-7)/0.03）。

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01XvmqDsZNyUo9P3sFQtoWVB

apps/api/src/auth/session.ts

@@ -72,14 +72,14 @@ export async function createSession(
     previousLogin,
     contestPasswords: {},
   }
-  await redis.set(
-    sessionKey(token),
-    JSON.stringify(value),
-    "EX",
-    config.sessionTtlSeconds,
-  )
-  await redis.sadd(userSessionsKey(userId), token)
-  await redis.expire(userSessionsKey(userId), config.sessionTtlSeconds)
+  // 三条写进一个 pipeline：一个班四十号人同时登录时，三趟往返和一趟的差别
+  // 全压在登录这一下上
+  await redis
+    .pipeline()
+    .set(sessionKey(token), JSON.stringify(value), "EX", config.sessionTtlSeconds)
+    .sadd(userSessionsKey(userId), token)
+    .expire(userSessionsKey(userId), config.sessionTtlSeconds)
+    .exec()
   setCookie(c, config.sessionCookie, token, {
     httpOnly: true,
     sameSite: "Lax",
@@ -193,10 +193,14 @@ async function getUserByToken(token: string | undefined): Promise<SessionResult>
     return { user: null, reason: "disabled" }
   }
 
-  await redis.expire(sessionKey(token), config.sessionTtlSeconds)
   // 反向索引跟着会话一起续期，否则活跃用户的索引会先于会话到期，
-  // 之后再吊销就找不到这张会话了
-  await redis.expire(userSessionsKey(session.userId), config.sessionTtlSeconds)
+  // 之后再吊销就找不到这张会话了。两条走一次 pipeline —— 这是全后端最热的 Redis
+  // 路径，每个带鉴权的请求都要走一趟，形状和 touchSession 里那对保持一致
+  await redis
+    .pipeline()
+    .expire(sessionKey(token), config.sessionTtlSeconds)
+    .expire(userSessionsKey(session.userId), config.sessionTtlSeconds)
+    .exec()
   // 唯一的收窄点。库里是 text 列，认不出来的值降成最低权限，见 toAdminType 的注释。
   return {
     user: {
@@ -232,13 +236,28 @@ export function readRequestSessionToken(request: Request) {
 /**
  * 会话还在就续期并返回 true，已登出或已过期返回 false。
  *
- * 用 EXPIRE 一条命令同时完成「判断存在」和「续期」，比 GET + EXPIRE 少一趟往返。
- * 续期这件事本身也是要的：HTTP 请求会走 getUserByToken 里的 redis.expire 续期，
+ * 用 EXPIRE 同时完成「判断存在」和「续期」，比 GET + EXPIRE 少一趟往返；两条 EXPIRE
+ * 走一次 pipeline，仍然只有一趟。
+ *
+ * 续期这件事本身是要的：HTTP 请求会走 getUserByToken 里的 redis.expire 续期，
  * 而只开着页面挂 WebSocket 的人一次请求都不发，不该因此被算成不活跃踢下线。
+ *
+ * **反向索引必须跟着一起续。** 走到这里的正是那种一次 HTTP 请求都不发的连接，
+ * 它碰不到 getUserByToken 里那两条并排的 expire。只续会话不续索引的话，索引先到期、
+ * 会话却被巡检一直续着，之后改密码 / 禁用账号走 revokeUserSessions 就 SMEMBERS
+ * 不到这张 token —— WebSocket 那边还有 publishSessionRevoked 按 userId 兜底能断掉，
+ * 但 HTTP 一侧拿着那张 cookie 照用不误，而改密码要的恰恰是让 HTTP 立刻失效。
  */
-export async function touchSession(token: string) {
+export async function touchSession(token: string, userId: number) {
   if (!token) return false
-  return (await redis.expire(sessionKey(token), config.sessionTtlSeconds)) === 1
+  const results = await redis
+    .pipeline()
+    .expire(sessionKey(token), config.sessionTtlSeconds)
+    .expire(userSessionsKey(userId), config.sessionTtlSeconds)
+    .exec()
+  // 索引那条的返回值不看：存量会话（反向索引上线之前签发的）本来就没有索引键，
+  // 续不到很正常，不能因此判定会话已死
+  return results?.[0]?.[1] === 1
 }
 
 async function getStoredSession(c: Context) {

apps/api/src/collab/handler.ts

@@ -202,7 +202,7 @@ export async function handleCollabMessage(ws: CollabSocket, raw: string) {
   }
 
   // 握手时校验过一次不算数 —— 这条连接能挂几个小时
-  if (!(await touchSession(ws.data.token))) {
+  if (!(await touchSession(ws.data.token, ws.data.userId))) {
     ws.close(1008, "Session expired")
     return
   }

apps/api/src/db/0011_user_lookup_indexes.sql

@@ -0,0 +1,2 @@
+CREATE INDEX "user_active_idx" ON "user" USING btree ("is_disabled","last_login" DESC NULLS FIRST);--> statement-breakpoint
+CREATE INDEX "user_class_name_idx" ON "user" USING btree ("class_name");

apps/api/src/db/meta/_journal.json

@@ -78,6 +78,13 @@
       "when": 1788409961010,
       "tag": "0010_fk_cascade_on_delete",
       "breakpoints": true
+    },
+    {
+      "idx": 11,
+      "version": "7",
+      "when": 1788788493497,
+      "tag": "0011_user_lookup_indexes",
+      "breakpoints": true
     }
   ]
 }

apps/api/src/db/schema.ts

@@ -664,6 +664,12 @@ export const user = pgTable("user", {
 	className: text("class_name"),
 }, (table) => [
 	unique("user_username_key").on(table.username),
+	// 「近两年登录过的活跃人数」—— problems/:id/beat-count 每次打开题目详情都要算一遍，
+	// 而这张表原来只有主键和 username 两个索引，那句统计是全表扫。
+	index("user_active_idx").using("btree", table.isDisabled.asc().nullsLast(), table.lastLogin.desc().nullsFirst()),
+	// 按班级 / 按年级（class_name like '241%'）取学生：班级榜、班级对比、AI 学情的
+	// 排名 scope 都走它，见 routes/classroom.ts 的 loadClassUsers。
+	index("user_class_name_idx").using("btree", table.className.asc().nullsLast()),
 ]);
 
 export const problemsetBadge = pgTable("problemset_badge", {

apps/api/src/redis.ts

@@ -2,18 +2,40 @@ import Redis from "ioredis"
 
 import { config } from "./config"
 
-export const redis = new Redis(config.redisUrl, {
-  maxRetriesPerRequest: 1,
-})
+/**
+ * 每条连接都要挂 error 监听。
+ *
+ * ioredis 对没有监听者的 error 走 silentEmit —— 不会像普通 EventEmitter 那样崩进程，
+ * 但会把连接错误直接 `console.error("[ioredis] Unhandled error event:", ...)` 打到
+ * stderr，绕开这里的日志，而且不说是哪条连接出的事。这个进程同时开着会话读写、
+ * 两条队列、一条订阅，出问题时「哪条」正是要先知道的。
+ */
+function withErrorLogging(client: Redis, name: string) {
+  client.on("error", (error) => {
+    console.error(`Redis connection error (${name})`, error)
+  })
+  return client
+}
+
+/**
+ * 会话、限流、发布事件都走这条。`maxRetriesPerRequest: 1` 是故意的：每个带鉴权的
+ * 请求都要读一次会话，Redis 不可用时快速失败成 500，比让请求挂在重试里更好。
+ */
+export const redis = withErrorLogging(
+  new Redis(config.redisUrl, { maxRetriesPerRequest: 1 }),
+  "main",
+)
 
 export function createBlockingRedis() {
-  return new Redis(config.redisUrl, {
-    maxRetriesPerRequest: null,
-  })
+  return withErrorLogging(
+    new Redis(config.redisUrl, { maxRetriesPerRequest: null }),
+    "blocking",
+  )
 }
 
 export function createSubscriberRedis() {
-  return new Redis(config.redisUrl, {
-    maxRetriesPerRequest: null,
-  })
+  return withErrorLogging(
+    new Redis(config.redisUrl, { maxRetriesPerRequest: null }),
+    "subscriber",
+  )
 }

apps/api/src/services/throttling.ts

@@ -61,6 +61,26 @@ redis.call('EXPIRE', key, ttl)
 return { allowed, tostring(wait) }
 `
 
+/**
+ * 注册成自定义命令而不是每次 `redis.eval`：eval 会把上面 900 多字节的脚本全文
+ * 一起发过去，而限流点在提交判题、AI 分析、流程图评分上，判题高峰期每条提交都要发
+ * 一遍。ioredis 的 defineCommand 走 EVALSHA，只发 40 字节的 sha1，遇到 NOSCRIPT
+ * 自动回退成一次 EVAL 把脚本重新灌进去 —— Redis 重启或 SCRIPT FLUSH 之后不用管。
+ */
+redis.defineCommand("throttleConsume", { numberOfKeys: 1, lua: CONSUME_SCRIPT })
+
+type ThrottleRedis = typeof redis & {
+  throttleConsume(
+    key: string,
+    capacity: string,
+    fillRate: string,
+    defaultCapacity: string,
+    now: string,
+    num: string,
+    ttl: string,
+  ): Promise<[number, string]>
+}
+
 function parseBucketConfig(value: unknown, fallback: BucketConfig): BucketConfig {
   if (!value || typeof value !== "object" || Array.isArray(value)) return fallback
   const raw = value as Record<string, unknown>
@@ -75,16 +95,38 @@ function parseBucketConfig(value: unknown, fallback: BucketConfig): BucketConfig
   }
 }
 
+/**
+ * 桶参数的进程内缓存。
+ *
+ * 限流点在提交判题、AI 分析、流程图评分上，原来每检查一次就查一次 `throttling`
+ * 配置项 —— 判题高峰期等于每条提交多一趟数据库，只为读一个几乎从不变的值。
+ * 上一代在 `options/options.py` 的 my_property 里也是带 TTL 缓存的，重写时漏掉了。
+ *
+ * 放进程内而不是 Redis：值只有几十字节，跨进程共享省不下什么，反倒要多一趟网络。
+ * `throttling` 没有后台界面，只能直接改库，改完最多一分钟后生效。
+ */
+const BUCKET_CACHE_TTL = 60_000
+const bucketCache = new Map<"user", { value: BucketConfig; expiresAt: number }>()
+
 export async function getBucketConfig(scope: "user"): Promise<BucketConfig> {
+  const cached = bucketCache.get(scope)
+  if (cached && cached.expiresAt > Date.now()) return cached.value
+
   const fallback = throttlingDefaults[scope]
+  let value: BucketConfig
   try {
     const values = await getOptions(["throttling"])
     const throttling = values.throttling
-    if (!throttling || typeof throttling !== "object" || Array.isArray(throttling)) return fallback
-    return parseBucketConfig((throttling as Record<string, unknown>)[scope], fallback)
+    value = !throttling || typeof throttling !== "object" || Array.isArray(throttling)
+      ? fallback
+      : parseBucketConfig((throttling as Record<string, unknown>)[scope], fallback)
   } catch {
+    // 读不到就退回默认值，但**不写缓存** —— 数据库抖一下不该让接下来一整分钟
+    // 全站都按默认参数限流
     return fallback
   }
+  bucketCache.set(scope, { value, expiresAt: Date.now() + BUCKET_CACHE_TTL })
+  return value
 }
 
 export type ConsumeResult = { allowed: true } | { allowed: false; wait: number }
@@ -99,9 +141,7 @@ export async function consumeToken(
   // 每次调用都会刷新 TTL，因此只有长时间无提交才会过期，届时桶早已回满，
   // 重新按 default_capacity 初始化只会更严，不会放水。
   const ttl = Math.ceil(bucket.capacity / bucket.fill_rate) + 60
-  const result = (await redis.eval(
-    CONSUME_SCRIPT,
-    1,
+  const result = await (redis as ThrottleRedis).throttleConsume(
     `throttling:${scope}:${identity}`,
     String(bucket.capacity),
     String(bucket.fill_rate),
@@ -109,7 +149,7 @@ export async function consumeToken(
     String(Date.now() / 1000),
     String(num),
     String(ttl),
-  )) as [number, string]
+  )
   if (Number(result[0]) === 1) return { allowed: true }
   return { allowed: false, wait: Number(result[1]) || 0 }
 }

apps/api/src/websocket.ts

@@ -178,7 +178,7 @@ export async function sweepSessions() {
     let alive = checked.get(token)
     if (alive === undefined) {
       try {
-        alive = await touchSession(token)
+        alive = await touchSession(token, ws.data.userId)
       } catch (error) {
         // Redis 抖一下不该把全班踢下线：这一轮直接放弃，下一轮再说
         console.error("Failed to verify websocket sessions", error)
@@ -299,7 +299,7 @@ async function handleMessage(
 
   // 会话可能在连接期间就失效了：用户在别的标签页登出，或者会话自己到期。
   // 握手时校验过一次不算数 —— 这条连接能挂几个小时。
-  if (!(await touchSession(ws.data.token))) {
+  if (!(await touchSession(ws.data.token, ws.data.userId))) {
     ws.close(1008, "Session expired")
     return
   }
@@ -438,9 +438,8 @@ export async function bridgeSubmissionEvents(
       console.error("Failed to bridge submission event", error)
     })
   })
-  subscriber.on("error", (error) => {
-    console.error("Submission event subscriber error", error)
-  })
+  // 连接层的 error 已经由 createSubscriberRedis 里的 withErrorLogging 打了
+  // （带连接名），这里再挂一个只会把同一条错误打两遍
   await subscriber.subscribe(
     submissionUpdateChannel,
     userEventChannel,

docker/clear-sessions.sh

@@ -0,0 +1,78 @@
+#!/usr/bin/env bash
+#
+# 一次性运维动作：清掉所有会话，强制全员重新登录。
+#
+# ## 为什么要跑
+#
+# 会话的反向索引 `user-sessions:<uid>` 是 498fc1c 才加的。在那之前签发的会话不在索引
+# 里，revokeUserSessions 靠 SMEMBERS 找不到它们 —— 也就是说**改密码、重置密码、禁用
+# 账号对这批会话统统无效**，只能等最长一个 SESSION_TTL_SECONDS（默认 7 天）自然过期。
+#
+# 学生密码是明文存着给老师查的，改密码正是密码泄露之后唯一的补救手段，这个空窗不能留。
+# 代价只是所有人重新登录一次。
+#
+# 跑过一次就不用再跑了：此后签发的会话都带索引。
+#
+# ## 两个站点都要跑
+#
+# 机房和服务器**共用一个数据库，但各有各的 Redis**（见 compose.school.yml 头部）。
+# 会话存在各自的 Redis 里，只清一边等于只解决一半。
+#
+# ## 为什么不是 FLUSHALL
+#
+# 同一个 Redis 里还装着 BullMQ 的判题队列（`bull:*`）。FLUSHALL 会把还在队列里的提交
+# 一起丢掉，那些 submission 会永远停在 PENDING，只能超管逐条重判 —— 而重判还会把
+# user_profile 的反范式计数带偏（见 apps/api/src/scripts/recount.ts）。
+# 这里只删 `session:*` 和 `user-sessions:*`。
+#
+# ## 用法
+#
+#   docker/clear-sessions.sh                      # 容器名默认 oj-redis
+#   CONTAINER=oj2-redis docker/clear-sessions.sh  # 本机 dev
+#   YES=1 docker/clear-sessions.sh                # 跳过确认
+#
+set -euo pipefail
+
+CONTAINER="${CONTAINER:-oj-redis}"
+
+say()  { printf '\n\033[1;36m==> %s\033[0m\n' "$*"; }
+ok()   { printf '    \033[32m✓\033[0m %s\n' "$*"; }
+die()  { printf '\n\033[1;31m❌ %s\033[0m\n\n' "$*" >&2; exit 1; }
+
+docker exec "$CONTAINER" redis-cli ping >/dev/null 2>&1 \
+  || die "连不上容器 $CONTAINER 里的 redis（用 CONTAINER=... 指定容器名）"
+
+# 用 SCAN 而不是 KEYS：KEYS 会阻塞住整个 Redis，而这上面还挂着判题队列和所有人的
+# 会话读写。xargs 分批是因为单条 DEL 传太多 key 会顶到命令行长度上限；每批 DEL 返回
+# 删掉的个数，累加起来就是总数。
+purge() {
+  docker exec "$CONTAINER" redis-cli --scan --pattern "$1" 2>/dev/null \
+    | xargs -r -n 400 docker exec "$CONTAINER" redis-cli del \
+    | awk '{ sum += $1 } END { print sum + 0 }'
+}
+
+count() { docker exec "$CONTAINER" redis-cli --scan --pattern "$1" 2>/dev/null | wc -l; }
+
+say "容器 $CONTAINER"
+before_sessions=$(count 'session:*')
+before_index=$(count 'user-sessions:*')
+before_bull=$(count 'bull:*')
+printf '    session:*        %s\n' "$before_sessions"
+printf '    user-sessions:*  %s\n' "$before_index"
+printf '    bull:*           %s（不动）\n' "$before_bull"
+
+if [ "${YES:-}" != "1" ]; then
+  read -rp $'\n    删掉上面的会话、让所有人重新登录？[y/N] ' answer
+  [ "$answer" = "y" ] || [ "$answer" = "Y" ] || die "已取消，什么都没做"
+fi
+
+say "清理"
+ok "session:*        删掉 $(purge 'session:*') 个"
+ok "user-sessions:*  删掉 $(purge 'user-sessions:*') 个"
+
+after_bull=$(count 'bull:*')
+[ "$after_bull" = "$before_bull" ] \
+  || die "判题队列的键数变了（$before_bull → $after_bull），这不该发生，请人工检查"
+ok "bull:* 仍是 $after_bull 个，判题队列没被动过"
+
+say "完成 —— 另一个站点的 Redis 也要跑一遍"