perf(worker-platform): 排行榜公共缓存与抢单短路,压制 8.5s 慢查询脉冲

- 排行榜:Top 10 改为按 period 全局共享缓存(TTL 30s),SQL 传 currentWorkerId=0;
  个人单数解耦为 assigned_worker_id 精准索引查询(TTL 15s),未上榜展示「暂未上榜」,
  消除「每个打手一份缓存」导致的 8.5 秒全表聚合穿透;
- 抢单短路:新增 work-order-grab-guard,对已关闭工单打 10s 进程内标记,命中直接 409;
  publishWorkOrderRealtimeChange 统一作废标记,避免撤单/超时重开/重新发布后误杀正常抢单;
- 我的订单:默认排序也走「先取 id 再批量回表」的延迟分页,避免宽列 material_json 参与排序;
- 文档:新增生产 CPU 波动分析报告,慢查询数字按 6,235 行日志实测重算,并标注修复状态与验收口径。

验证:typecheck / check:sql / eslint / prettier 通过;后端 npm test 914 项(910 pass、4 skip、0 fail)。
This commit is contained in:
yml2213
2026-10-30 22:42:11 +08:00
parent 4ef52f11ba
commit 44791a9c8b
9 changed files with 421 additions and 25 deletions
@@ -435,12 +435,11 @@ export async function listWorkOrders({
const effectiveOrderBy = pinnedFirst ? `wo.pinned_at DESC NULLS LAST, ${orderBy}` : orderBy
params.push(pageSize, offset)
// 打手端「我的订单」的排序是表达式(依赖打手参数与拼单表),**没有索引能支撑**,
// 规划器只能把该打手全部可见单连同宽列(含 1.2KB 的 material_json)物化后排序。
// 生产实测:名下 6600 单的打手首访 1112ms、每次读约 60MB(列表缓存 2 秒一失效,
// 所以这个代价每分钟会重复付好几次)。这里改成"先只取当前页的 id,再按 id 取详情":
// 排序阶段只带窄列,宽列只回表 20 行。语义(行集、顺序、总数)完全不变。
if (sort === 'worker_claimed_at_desc' && workerSharingId && !pinnedFirst) {
// 打手端「我的订单」:不论是按接单时间排序还是按 id 默认排序,
// 都走"先只取当前页的 id,再按 id 批量取详情"的延迟分页模式。
// 避免宽列(含 1.2KB 的 material_json)在匹配和排序时全表物化。
// 语义(行集、顺序、总数)完全不变,大幅降低内存与数据库缓冲读。
if (workerSharingId && !pinnedFirst) {
const idsResult = await query<{ id: number }>(
`SELECT wo.id
FROM work_orders wo
@@ -29,6 +29,62 @@ export async function countWorkerAcceptedOrders(workerId: number | string): Prom
return Number(result.rows[0]?.total || 0)
}
/** 按周期轻量统计单打手完成单数(走 assigned_worker_id 精准索引,耗时 1~2ms,避免全表扫描) */
export async function countWorkerAcceptedOrdersByPeriod(
workerId: number | string,
period: 'all' | 'daily' = 'all',
): Promise<number> {
const normalizedWorkerId = Number(workerId)
if (!normalizedWorkerId) return 0
const dailyOnly = period === 'daily'
const result = await query<{ total: number }>(
`
WITH leaderboard_day AS (
SELECT
date_trunc('day', CURRENT_TIMESTAMP AT TIME ZONE 'Asia/Shanghai')
AT TIME ZONE 'Asia/Shanghai' AS starts_at
)
SELECT COUNT(DISTINCT order_key)::int AS total
FROM (
SELECT
COALESCE(
'order:' || NULLIF(wo.order_id::text, ''),
'platform_order:' || NULLIF(BTRIM(wo.platform_order_id), ''),
'work_order:' || wo.id::text
) AS order_key
FROM work_orders wo
CROSS JOIN leaderboard_day
WHERE wo.assigned_worker_id = $1
AND wo.status = 'accepted'
AND (
$2::boolean = false
OR (wo.accepted_at >= leaderboard_day.starts_at AND wo.accepted_at < leaderboard_day.starts_at + INTERVAL '1 day')
)
UNION
SELECT
COALESCE(
'order:' || NULLIF(wo.order_id::text, ''),
'platform_order:' || NULLIF(BTRIM(wo.platform_order_id), ''),
'work_order:' || wo.id::text
) AS order_key
FROM work_order_shares wos
INNER JOIN work_orders wo ON wo.id = wos.work_order_id
CROSS JOIN leaderboard_day
WHERE wos.worker_id = $1
AND wos.status = 'accepted'
AND (
$2::boolean = false
OR (wos.accepted_at >= leaderboard_day.starts_at AND wos.accepted_at < leaderboard_day.starts_at + INTERVAL '1 day')
)
) AS completed_orders
`,
[normalizedWorkerId, dailyOnly],
)
return Number(result.rows[0]?.total || 0)
}
export async function listWorkerCompletionLeaderboard(input: {
limit: number
currentWorkerId: number
@@ -38,6 +38,7 @@ import {
resolveWorkerPermissions,
stripWorkerPersonalShareFields,
} from '../worker-platform/mappers.js'
import { evictWorkOrderGrabBlock } from '../worker-platform/work-order-grab-guard.js'
import { invalidateWorkerMyOrderCaches } from '../worker-platform/worker-my-order-cache.js'
export type RealtimeEventType =
@@ -188,6 +189,9 @@ export function publishWorkOrderRealtimeChange(input: {
hallChanged?: boolean
}) {
const workOrderId = Math.max(0, Number(input.workOrderId) || 0)
// 工单状态变更(撤单 / 超时重开 / 重新发布)会复用同一个工单 id,
// 必须立刻作废抢单短路标记,否则正常打手会在 TTL 内被误判 409。
evictWorkOrderGrabBlock(workOrderId)
void publishWorkOrderSnapshots({ ...input, workOrderId }).catch(() => {
publishRealtimeEvent({
type: 'work_order.changed',
@@ -0,0 +1,58 @@
import assert from 'node:assert/strict'
import { readFile } from 'node:fs/promises'
import test from 'node:test'
import {
blockWorkOrderGrab,
evictWorkOrderGrabBlock,
isWorkOrderGrabBlocked,
resetWorkOrderGrabGuard,
} from './work-order-grab-guard.js'
test('抢单短路标记在 TTL 内命中,过期后自动失效', () => {
resetWorkOrderGrabGuard()
const now = 1_000_000
blockWorkOrderGrab('195657', now)
assert.equal(isWorkOrderGrabBlocked('195657', now + 1), true, 'TTL 内必须短路')
assert.equal(isWorkOrderGrabBlocked(195657, now + 9_999), true, '数字/字符串 id 必须等价')
assert.equal(isWorkOrderGrabBlocked('195657', now + 10_000), false, 'TTL 到期必须放行回查数据库')
assert.equal(isWorkOrderGrabBlocked('195657', now + 10_001), false)
})
test('工单重新开放(撤单/超时重开/重新发布)必须立即作废短路标记', () => {
resetWorkOrderGrabGuard()
blockWorkOrderGrab('195657', 5_000)
evictWorkOrderGrabBlock('195657')
assert.equal(
isWorkOrderGrabBlocked('195657', 5_001),
false,
'标记不失效会让正常打手在 TTL 内被误判 409',
)
})
test('空 id 不参与短路,重复打标记只顺延 TTL', () => {
resetWorkOrderGrabGuard()
blockWorkOrderGrab('')
assert.equal(isWorkOrderGrabBlocked(''), false)
blockWorkOrderGrab('42', 0)
blockWorkOrderGrab('42', 9_000)
assert.equal(isWorkOrderGrabBlocked('42', 18_999), true, '第二次命中应顺延到 19000ms')
assert.equal(isWorkOrderGrabBlocked('42', 19_000), false)
})
test('短路标记必须在抢单入口打上、在实时变更出口失效', async () => {
const [hallActionSource, realtimeSource] = await Promise.all([
readFile(new URL('./worker-hall-action-service.ts', import.meta.url), 'utf8'),
readFile(new URL('../realtime/realtime-event-service.ts', import.meta.url), 'utf8'),
])
assert.ok(
hallActionSource.includes('isWorkOrderGrabBlocked(') &&
hallActionSource.includes('blockWorkOrderGrab('),
'抢单入口必须使用 work-order-grab-guard 的短路判定与打标记',
)
assert.ok(
realtimeSource.includes('evictWorkOrderGrabBlock('),
'publishWorkOrderRealtimeChange 必须作废短路标记,否则工单重新开放后会被误判 409',
)
})
@@ -0,0 +1,55 @@
/**
* 抢单短路标记:被脚本/外挂死循环狂刷的「已关闭工单」在进程内存里直接拒绝,
* 不再穿透 worker_users / work_orders 查询与行级锁探测。
*
* 唯一要防的副作用是「工单重新开放」:撤单、超时重开、重新发布都会复用同一个
* 工单 id。如果标记不失效,正常打手会在 TTL 内被误判成 409。
* 因此所有会把工单重新放回大厅的路径都必须调用 evictWorkOrderGrabBlock
* (目前统一挂在 publishWorkOrderRealtimeChange 上,它正是这些路径的共同出口)。
*/
const NON_OPEN_TTL_MS = 10_000
const MAX_ENTRIES = 2_000
const expiresAtByWorkOrderId = new Map<string, number>()
function normalizeWorkOrderId(workOrderId: number | string): string {
return String(workOrderId ?? '').trim()
}
/** 命中未过期的短路标记时返回 true,调用方据此直接抛 409,不查库。 */
export function isWorkOrderGrabBlocked(workOrderId: number | string, now = Date.now()): boolean {
const key = normalizeWorkOrderId(workOrderId)
if (!key) return false
const expiresAt = expiresAtByWorkOrderId.get(key)
if (!expiresAt) return false
if (expiresAt <= now) {
expiresAtByWorkOrderId.delete(key)
return false
}
return true
}
/** 工单确认处于不可抢状态后打标记;同一工单重复命中只顺延 TTL。 */
export function blockWorkOrderGrab(workOrderId: number | string, now = Date.now()): void {
const key = normalizeWorkOrderId(workOrderId)
if (!key) return
if (expiresAtByWorkOrderId.size >= MAX_ENTRIES) {
for (const [entryKey, expiresAt] of expiresAtByWorkOrderId) {
if (expiresAt <= now) expiresAtByWorkOrderId.delete(entryKey)
}
// 极端情况下(短时间内 2000 个不同工单被判死)整体清空,宁可少一次短路也不能无限涨内存。
if (expiresAtByWorkOrderId.size >= MAX_ENTRIES) expiresAtByWorkOrderId.clear()
}
expiresAtByWorkOrderId.set(key, now + NON_OPEN_TTL_MS)
}
/** 工单状态发生变更(重新开放/重新发布/被取消)时立即作废标记。 */
export function evictWorkOrderGrabBlock(workOrderId: number | string): void {
const key = normalizeWorkOrderId(workOrderId)
if (key) expiresAtByWorkOrderId.delete(key)
}
/** 仅供测试重置进程内状态。 */
export function resetWorkOrderGrabGuard(): void {
expiresAtByWorkOrderId.clear()
}
@@ -19,6 +19,7 @@ import {
} from './mappers.js'
import { getWorkerHallConfig } from './worker-hall-config-service.js'
import { invalidateWorkerHallListCache } from './worker-hall-service.js'
import { blockWorkOrderGrab, isWorkOrderGrabBlocked } from './work-order-grab-guard.js'
import { publishWorkOrderParticipantChange } from './worker-order-realtime-service.js'
import {
getRequiredWorkOrder,
@@ -31,12 +32,24 @@ import {
publishWorkOrderRealtimeChange,
} from '../realtime/realtime-event-service.js'
// 已被抢/已关闭工单的高频抢单短路:标记与失效逻辑放在 work-order-grab-guard,
// 由实时变更出口(publishWorkOrderRealtimeChange)在工单重新开放时立即作废,
// 避免撤单/超时重开后正常打手被 TTL 内的残留标记误判 409。
export async function grabWorkerHallOrder(
workOrderId: number | string,
session: WorkerSession,
payload: JsonObject = {},
) {
requireActiveWorkerSession(session)
const nowMs = Date.now()
if (isWorkOrderGrabBlocked(workOrderId, nowMs)) {
throw createHttpError('该订单已不可抢', {
statusCode: 409,
errorCode: 'work_order_not_open',
})
}
const worker = await getRequiredWorker(session.workerId)
const permissions = resolveWorkerPermissions(worker)
const hallConfig = getWorkerHallConfig()
@@ -47,6 +60,7 @@ export async function grabWorkerHallOrder(
visibleDelaySeconds > 0 ? new Date(Date.now() - visibleDelaySeconds * 1000).toISOString() : ''
const workOrder = await getRequiredWorkOrder(workOrderId)
if (workOrder.status !== WORK_ORDER_STATUS.OPEN || workOrder.assigned_worker_id) {
blockWorkOrderGrab(workOrderId, nowMs)
throw createHttpError('该订单已不可抢', {
statusCode: 409,
errorCode: 'work_order_not_open',
@@ -1,5 +1,6 @@
import { WORK_ORDER_STATUS } from '../../domain/work-order-status.js'
import {
countWorkerAcceptedOrdersByPeriod,
listWorkCategories,
listWorkerCompletionLeaderboard,
listWorkOrderSharesByOrderIds,
@@ -35,14 +36,12 @@ const hallListCache = new WorkerListCache(HALL_LIST_CACHE_TTL_MS)
*/
const HALL_MAX_PAGE_SIZE = 20
// 排行榜每次都要对全量 work_orders + work_order_shares 做去重聚合(生产实测
// 742ms / 35,483 块 ≈ 277MB),而结果只在工单验收完成时才变化。
// 大厅每次刷新/窗口聚焦都会触发前端重新拉取,不缓存时该查询独占全库约 8% 的缓冲读。
// TTL 取 30s:对排行榜而言足够新鲜(名次变化以分钟计),又能把重算频率压掉两个数量级。
// 注意:结果里包含"当前打手自己的名次"(SQL 里 OR worker_id = $2),缓存键必须带上
// workerId,否则会把 A 的名次错发给 B。
// 排行榜全局公共缓存:Top 10 对全站所有人相同,公共数据全局共享同一份缓存(TTL 30s)。
// 彻底消解按 workerId 区分缓存导致的缓存击穿与单次 8.5 秒全表扫描问题。
// 打手自身若在 Top 10 中直接取其 rank,若未上榜则名次为 null,单数走 1.8ms 精准索引独立缓存。
const HALL_LEADERBOARD_CACHE_TTL_MS = 30_000
const hallLeaderboardCache = new WorkerListCache(HALL_LEADERBOARD_CACHE_TTL_MS)
const workerPersonalCountCache = new WorkerListCache(15_000)
export async function listWorkerHallOrders(query: JsonObject = {}, session: WorkerSession) {
requireActiveWorkerSession(session)
@@ -154,15 +153,32 @@ export async function getWorkerHallCategories(session: WorkerSession) {
export async function getWorkerHallLeaderboard(session: WorkerSession, period: unknown = 'all') {
requireActiveWorkerSession(session)
const leaderboardPeriod = period === 'daily' ? 'daily' : 'all'
const cacheKey = JSON.stringify([leaderboardPeriod, Number(session.workerId)])
// 全局公共缓存 Key,所有打手共享同一份 Top 10,TTL 30s
const cacheKey = JSON.stringify([leaderboardPeriod])
const entries = await hallLeaderboardCache.getOrSet(cacheKey, () =>
listWorkerCompletionLeaderboard({
limit: 10,
currentWorkerId: session.workerId,
currentWorkerId: 0,
period: leaderboardPeriod,
}),
)
const current = entries.find((entry) => entry.workerId === session.workerId)
const currentWorkerId = Number(session.workerId)
const topEntry = entries.find((entry) => entry.workerId === currentWorkerId)
let myRank: number | null = null
let myCount = 0
if (topEntry) {
myRank = topEntry.rank
myCount = topEntry.acceptedOrderCount
} else if (currentWorkerId) {
// 未上榜打手:名次为 null(前端显示为“未上榜”),单数走 1.8ms 精准索引查询,带 15s 短缓存
const myCountKey = JSON.stringify([leaderboardPeriod, currentWorkerId])
myCount = await workerPersonalCountCache.getOrSet(myCountKey, () =>
countWorkerAcceptedOrdersByPeriod(currentWorkerId, leaderboardPeriod),
)
}
return {
items: entries
.filter((entry) => entry.rank <= 10)
@@ -171,8 +187,8 @@ export async function getWorkerHallLeaderboard(session: WorkerSession, period: u
displayName: maskWorkerLeaderboardDisplayName(entry.displayName),
})),
current: {
rank: current?.rank || null,
acceptedOrderCount: current?.acceptedOrderCount || 0,
rank: myRank,
acceptedOrderCount: myCount,
},
}
}
@@ -3285,8 +3285,8 @@ test('打手端「我的订单」必须走"先取 id 再取详情"的延迟分
'utf8',
)
assert.ok(
src.includes("sort === 'worker_claimed_at_desc' && workerSharingId && !pinnedFirst"),
'缺少打手排序的延迟分页分支:该排序是表达式,无索引可支撑,会物化全部可见单的宽列',
src.includes('if (workerSharingId && !pinnedFirst) {'),
'缺少打手延迟分页分支:按接单时间与默认 id 排序都必须先取窄列 id,避免宽列全量物化',
)
assert.ok(
src.includes('SELECT wo.id'),
@@ -3372,17 +3372,17 @@ test('接单排行榜必须有短缓存(单次全表聚合 277MB/742ms,独
'排行榜必须走 getOrSet(同时合并并发请求)',
)
// 结果包含"当前打手自己的名次"(SQL 里 OR worker_id = $2),
// 缓存键漏了 workerId 会把 A 的名次错发给 B —— 这是正确性问题,不是性能问题。
// 公共 Top 10 全局共享同一份缓存(Key 仅为 period),
// 个人名次与单数解耦计算,彻底杜绝把 workerId 放入 Top 10 缓存导致的并发全表扫描击穿。
const leaderboardBody = src.slice(
src.indexOf('export async function getWorkerHallLeaderboard'),
src.indexOf('export function maskWorkerLeaderboardDisplayName'),
)
assert.ok(
leaderboardBody.includes('session.workerId') &&
/cacheKey/.test(leaderboardBody) &&
leaderboardBody.includes('JSON.stringify'),
'排行榜缓存键必须包含 workerId 与 period',
/cacheKey/.test(leaderboardBody) &&
leaderboardBody.includes('JSON.stringify([leaderboardPeriod])') &&
leaderboardBody.includes('workerPersonalCountCache'),
'排行榜公共 Top 10 缓存键必须为全局共享的 period,个人单数必须解耦缓存',
)
})