yml2213
f0d745ea50
## 一、刚下的封禁会被对账撤销（生产实测） Tier 3 是「先把 tracker.banned 置 true，再 await 写库」。在写库提交前的窗口里， 并发的另一个请求会走对账分支、读到 status 仍是 active，于是把刚下的封禁撤销掉。 生产日志（worker 28）： 17:20:11.361 WARN 封禁标记与数据库不一致，已自动解除并放行 ← 对账把刚下的封禁撤了 17:20:11.388 ERROR 打手账号已被系统自动永久封禁！ count=30 ← 27ms 后才写出封禁 后果：刷子能漏进来一个请求；日志还会让运营以为"数据库里已经解封了"。 修法：新增 bannedAt，刚下的封禁在 BAN_GRACE_MS(60s) 内不与数据库对账。 人工解封走 releaseWorkerRiskTracker() 直接删 tracker，不受宽限期影响，处置不会变慢。 ## 二、账号不存在时会写下"已永久封禁"的假记录 updateWorkerUser 在账号不存在时返回 null 而**不抛错**，原代码直接把返回值丢弃， 于是照样打出"已被系统自动永久封禁"的日志并写 tier3_autofrozen 留痕。 外挂伪造成随机 uid 时就会命中这条路径，留下一批查不到的假封禁记录。 修法：显式判断返回值，为空则记明确错误并跳过留痕。 ## 测试 新增 2 个用例： - 刚下的封禁在宽限期内不得被对账撤销（锁住上面的竞态） - 账号不存在时不写"已封禁"的假日志 并更新 2 个原有用例：宽限期内本就不发起对账，原断言（bannedVerifiedAt > 0）已过时。 风控测试 11/11；后端全量 559 测试 556 通过 0 失败。
524 lines
19 KiB
TypeScript
524 lines
19 KiB
TypeScript
import type { NextFunction, Request, Response } from 'express'
|
||
import { buildErrorPayload, createHttpError } from '../utils/http.js'
|
||
import { logError, logWarn } from '../utils/logger.js'
|
||
import { nowIso } from '../utils/time.js'
|
||
import { getWorkerUserById, updateWorkerUser } from '../repositories/worker-platform/worker-repo.js'
|
||
import { revokeAllWorkerSessions } from '../repositories/worker-platform/worker-session-repo.js'
|
||
import { insertWorkerRiskEvent } from '../repositories/worker-platform/worker-risk-repo.js'
|
||
import { getWorkerRiskConfig } from '../services/worker-platform/worker-risk-config-service.js'
|
||
|
||
/**
|
||
* 打手端阶梯式风控配置与状态机
|
||
*
|
||
* 针对外挂/脚本毫秒级死循环狂刷接单大厅、挤占单线程 Node 事件循环与数据库连接池的场景：
|
||
* Tier 1 (平滑限流): 窗口内超过 N 次请求 -> 429 快速拒绝，避免穿透后续 DB
|
||
* Tier 2 (风控强化): 违规窗口内累计触发 M 次 -> 进入静默惩罚期，秒拒且带 Retry-After
|
||
* Tier 3 (恶意封号): 在惩罚期内仍不停止、累计违规达 K 次 -> 自动封禁账号并作废 session
|
||
*
|
||
* 阈值全部来自 worker-risk-config-service（后台可调，保存即热生效），不再硬编码。
|
||
*
|
||
* 状态仍然是**进程内存**（重启清零），这是刻意的：按请求更新内存计数是纳秒级操作，
|
||
* 落库会瞬间打爆数据库。内存状态通过 listActiveWorkerRiskTrackers() 暴露给后台实时展示；
|
||
* 只有「显著事件」（进入二级惩罚、三级封号、人工处置）才异步写 worker_risk_events 留痕。
|
||
*/
|
||
|
||
interface WorkerRiskTracker {
|
||
// 当前滑动窗口内请求计数
|
||
windowCount: number
|
||
windowResetAt: number
|
||
// 短期累计违规次数 (429 触发次数)
|
||
violationCount: number
|
||
violationResetAt: number
|
||
// Tier 2 惩罚截止时间戳 (0 表示正常)
|
||
penalizedUntil: number
|
||
// 是否已被 Tier 3 自动封禁
|
||
banned: boolean
|
||
// 本进程把 banned 置为 true 的时刻。用于给「刚下的封禁」一个宽限期，
|
||
// 避免对账读未提交把刚写的封禁撤销（见 isWorkerStillFrozen）。
|
||
bannedAt: number
|
||
// 上次与数据库对账「是否仍处于 frozen」的时间戳，0 表示尚未对账
|
||
bannedVerifiedAt: number
|
||
// 便于后台展示的追溯信息（最近一次见到的身份/IP 与时间）
|
||
lastWorkerId: number | null
|
||
lastIp: string
|
||
lastSeenAt: number
|
||
// 累计被拒请求数（仅用于展示，不落库）
|
||
rejectedCount: number
|
||
}
|
||
|
||
const riskTrackers = new Map<string, WorkerRiskTracker>()
|
||
let lastCleanupAt = 0
|
||
|
||
/**
|
||
* 内存封禁标记与数据库的对账间隔。
|
||
*
|
||
* 为什么必须有：`banned` 只是数据库 `worker_users.status === 'frozen'` 的缓存。
|
||
* 若把它当成永久事实，后台把账号改回 active 后，中间件仍会在鉴权之前直接 403
|
||
* ——数据库里已经是正常账号，人却永远接不了单，只能等容器重启。
|
||
* 因此这里按 TTL 回查数据库；数据库是唯一事实来源。
|
||
*
|
||
* 代价可控：只有被判定为 banned 的 key 才会走到这里，且每个 key 最多每 30 秒查一次。
|
||
* 即便该账号正以 65 QPS 持续刷接口，对数据库也只是每 30 秒一次查询。
|
||
*/
|
||
const BAN_RECHECK_INTERVAL_MS = 30_000
|
||
|
||
/**
|
||
* 刚下的封禁在多久内不与数据库对账。
|
||
*
|
||
* 必需：Tier 3 是「先把 tracker.banned 置 true，再 await 写库」。在写库提交之前的这个窗口里，
|
||
* 并发的另一个请求会走对账分支，读到 status 仍是 active，于是把刚刚下的封禁撤销掉
|
||
* ——生产实测出现过 27ms 的窗口（17:20:11.361 判定"不一致"、.388 才写出封禁），
|
||
* 结果是刷子能漏进来一个请求，日志还会误导运营。
|
||
*
|
||
* 前台解封（后台复核 / 风控页解封）走的是 releaseWorkerRiskTracker()，直接删 tracker，
|
||
* 不受这个宽限期影响，所以不会拖慢人工处置。
|
||
*/
|
||
const BAN_GRACE_MS = 60_000
|
||
|
||
/**
|
||
* 查账号状态的函数签名。抽成类型 + 可注入，是为了让「数据库已解封、内存仍封禁」
|
||
* 这条修复路径能被测试覆盖（测试环境没有 Postgres）。生产不传即用真实仓储。
|
||
*/
|
||
export type WorkerRiskWorkerLookup = (workerId: number) => Promise<{ status?: string } | null>
|
||
|
||
/** 后台展示用的 tracker 快照上限，避免极端情况下返回超大数组 */
|
||
const TRACKER_SNAPSHOT_LIMIT = 200
|
||
|
||
/** 从 Authorization 请求头极速无 DB 提取 workerId (JWT payload base64 解码，微秒级) */
|
||
export function extractFastWorkerId(req: Request): number | null {
|
||
const auth = String(req.headers.authorization || '').trim()
|
||
if (!auth.startsWith('Bearer ') && !auth.startsWith('bearer ')) {
|
||
return null
|
||
}
|
||
const token = auth.slice(7).trim()
|
||
const dotIndex = token.indexOf('.')
|
||
if (dotIndex <= 0) return null
|
||
const encodedPayload = token.slice(0, dotIndex)
|
||
try {
|
||
const raw = Buffer.from(encodedPayload, 'base64url').toString('utf8')
|
||
const payload = JSON.parse(raw) as { typ?: string; uid?: number | string }
|
||
if (payload.typ === 'worker' && payload.uid) {
|
||
const id = Number(payload.uid)
|
||
return id > 0 ? id : null
|
||
}
|
||
} catch {
|
||
// 忽略异常，降级按 IP 识别
|
||
}
|
||
return null
|
||
}
|
||
|
||
export function resolveClientIp(req: Request): string {
|
||
return String(
|
||
req.ip ||
|
||
String(req.headers['x-forwarded-for'] || '').split(',')[0] ||
|
||
req.socket?.remoteAddress ||
|
||
'unknown',
|
||
).trim()
|
||
}
|
||
|
||
/** 获取跟踪目标 key：优先以 workerId 维度（防止切 IP），兜底以 IP 维度（防止未登录或换账号） */
|
||
function getTargetKeys(req: Request): { primaryKey: string; workerId: number | null; ip: string } {
|
||
const ip = resolveClientIp(req)
|
||
const workerId = extractFastWorkerId(req)
|
||
const primaryKey = workerId ? `worker:${workerId}` : `ip:${ip}`
|
||
return { primaryKey, workerId, ip }
|
||
}
|
||
|
||
function getOrCreateTracker(key: string, now: number): WorkerRiskTracker {
|
||
let tracker = riskTrackers.get(key)
|
||
if (!tracker) {
|
||
const config = getWorkerRiskConfig()
|
||
tracker = {
|
||
windowCount: 0,
|
||
windowResetAt: now + config.tier1WindowMs,
|
||
violationCount: 0,
|
||
violationResetAt: now + config.violationWindowMs,
|
||
penalizedUntil: 0,
|
||
banned: false,
|
||
bannedAt: 0,
|
||
bannedVerifiedAt: 0,
|
||
lastWorkerId: null,
|
||
lastIp: '',
|
||
lastSeenAt: now,
|
||
rejectedCount: 0,
|
||
}
|
||
riskTrackers.set(key, tracker)
|
||
}
|
||
return tracker
|
||
}
|
||
|
||
/**
|
||
* 判断仍被封禁的 key 是否真的还该被封。
|
||
*
|
||
* 返回 true 表示维持封禁；返回 false 表示数据库侧已解封（或账号已不存在），
|
||
* 调用方应清掉内存标记并放行。
|
||
*
|
||
* 查询失败时**保守维持封禁**——风控宁可误挡，也不能因为一次数据库抖动把刷子放进来。
|
||
*/
|
||
async function isWorkerStillFrozen(
|
||
tracker: WorkerRiskTracker,
|
||
now: number,
|
||
lookupWorker: WorkerRiskWorkerLookup,
|
||
): Promise<boolean> {
|
||
// 刚下的封禁先给一段宽限期：此刻数据库写入可能还没提交，
|
||
// 立刻对账会把刚封的账号又放出去（读未提交竞态）。
|
||
if (tracker.bannedAt && now - tracker.bannedAt < BAN_GRACE_MS) {
|
||
return true
|
||
}
|
||
|
||
if (tracker.bannedVerifiedAt && now - tracker.bannedVerifiedAt < BAN_RECHECK_INTERVAL_MS) {
|
||
return true
|
||
}
|
||
tracker.bannedVerifiedAt = now
|
||
|
||
// banned 只会由 Tier 3 设置，而 Tier 3 要求能解出 workerId，因此这里必然有 workerId；
|
||
// 兜底返回 true 以免出现无法校验却放行的情况。
|
||
const workerId = tracker.lastWorkerId
|
||
if (!workerId) return true
|
||
|
||
try {
|
||
const worker = await lookupWorker(workerId)
|
||
if (!worker) {
|
||
// 账号已被删除，封禁自然失效
|
||
tracker.banned = false
|
||
tracker.bannedAt = 0
|
||
return false
|
||
}
|
||
if (worker.status === 'frozen') return true
|
||
|
||
// 解封后必须把整轮窗口状态一起清零，只清 banned 是不够的：
|
||
// 上一轮刷出来的 windowCount / violationCount 仍在窗口内，账号刚恢复就会立刻
|
||
// 被 Tier 1 限流、并在 60 秒内再次凑满 Tier 2 触发条件而二次封禁。
|
||
tracker.banned = false
|
||
tracker.bannedAt = 0
|
||
tracker.penalizedUntil = 0
|
||
tracker.violationCount = 0
|
||
tracker.violationResetAt = 0
|
||
tracker.windowCount = 0
|
||
tracker.windowResetAt = 0
|
||
tracker.rejectedCount = 0
|
||
return false
|
||
} catch (err) {
|
||
logError('[worker/risk-control]', '封禁状态对账失败，暂时维持封禁', { workerId, err })
|
||
return true
|
||
}
|
||
}
|
||
|
||
/** 接单大厅与高频打手接口的风控拦截中间件 */
|
||
export function createWorkerHallRiskMiddleware(
|
||
deps: { lookupWorker?: WorkerRiskWorkerLookup } = {},
|
||
) {
|
||
const lookupWorker: WorkerRiskWorkerLookup = deps.lookupWorker ?? getWorkerUserById
|
||
return async (req: Request, res: Response, next: NextFunction): Promise<void> => {
|
||
const config = getWorkerRiskConfig()
|
||
if (!config.enabled) {
|
||
next()
|
||
return
|
||
}
|
||
|
||
const now = Date.now()
|
||
cleanupExpiredTrackers(now)
|
||
|
||
const { primaryKey, workerId, ip } = getTargetKeys(req)
|
||
const tracker = getOrCreateTracker(primaryKey, now)
|
||
if (workerId) tracker.lastWorkerId = workerId
|
||
if (ip) tracker.lastIp = ip
|
||
tracker.lastSeenAt = now
|
||
|
||
// 1. 被封号则拦截。但必须先与数据库对账：`banned` 只是数据库 status='frozen' 的缓存，
|
||
// 后台解封（或直接改库）后内存标记若不失效，账号在数据库里已是 active、
|
||
// 人却仍然接不了单——生产上已实际发生，只能等容器重启才恢复。
|
||
if (tracker.banned) {
|
||
if (await isWorkerStillFrozen(tracker, now, lookupWorker)) {
|
||
tracker.rejectedCount += 1
|
||
const error = createHttpError('接单账号已被系统安全风控封禁', {
|
||
statusCode: 403,
|
||
errorCode: 'worker_account_frozen',
|
||
})
|
||
res.status(403).json(buildErrorPayload(error, '接单账号已被封禁'))
|
||
return
|
||
}
|
||
logWarn('[worker/risk-control]', '封禁标记与数据库不一致，已自动解除并放行', {
|
||
primaryKey,
|
||
workerId,
|
||
})
|
||
}
|
||
|
||
// 2. 检查是否处于 Tier 2 惩罚期
|
||
if (tracker.penalizedUntil > now) {
|
||
tracker.violationCount += 1
|
||
tracker.rejectedCount += 1
|
||
const retryAfterSeconds = Math.max(1, Math.ceil((tracker.penalizedUntil - now) / 1000))
|
||
res.setHeader('Retry-After', String(retryAfterSeconds))
|
||
|
||
// 检查是否升级到 Tier 3 自动封号
|
||
if (tracker.violationCount >= config.tier3TriggerViolations && workerId && !tracker.banned) {
|
||
tracker.banned = true
|
||
tracker.bannedAt = now
|
||
await executeAutoFreezeWorker(workerId, ip, tracker.violationCount)
|
||
const error = createHttpError('恶意高频刷接口多次不改，接单账号已被系统自动永久封禁', {
|
||
statusCode: 403,
|
||
errorCode: 'worker_account_frozen',
|
||
})
|
||
res.status(403).json(buildErrorPayload(error, '接单账号已被永久封禁'))
|
||
return
|
||
}
|
||
|
||
logWarn('[worker/risk-control]', '打手处于二级风控惩罚期，拒绝请求', {
|
||
primaryKey,
|
||
workerId,
|
||
ip,
|
||
retryAfterSeconds,
|
||
violationCount: tracker.violationCount,
|
||
})
|
||
|
||
const error = createHttpError('系统检测到异常高频请求，已触发安全风控，请稍后再试', {
|
||
statusCode: 429,
|
||
errorCode: 'risk_penalized',
|
||
})
|
||
res.status(429).json(buildErrorPayload(error, '系统安全风控限制中'))
|
||
return
|
||
}
|
||
|
||
// 3. 正常滑动窗口限流检查 (Tier 1)
|
||
if (now > tracker.windowResetAt) {
|
||
tracker.windowCount = 0
|
||
tracker.windowResetAt = now + config.tier1WindowMs
|
||
}
|
||
tracker.windowCount += 1
|
||
|
||
if (tracker.windowCount <= config.tier1MaxRequests) {
|
||
next()
|
||
return
|
||
}
|
||
|
||
// 4. 触发 Tier 1 超限：记录违规
|
||
tracker.rejectedCount += 1
|
||
if (now > tracker.violationResetAt) {
|
||
tracker.violationCount = 0
|
||
tracker.violationResetAt = now + config.violationWindowMs
|
||
}
|
||
tracker.violationCount += 1
|
||
|
||
// 5. 检查是否升级为 Tier 2 惩罚
|
||
if (tracker.violationCount >= config.tier2TriggerViolations) {
|
||
tracker.penalizedUntil = now + config.tier2PenaltyMs
|
||
const retryAfter = Math.ceil(config.tier2PenaltyMs / 1000)
|
||
res.setHeader('Retry-After', String(retryAfter))
|
||
|
||
const penaltyMinutes = Math.round(config.tier2PenaltyMs / 60_000)
|
||
logWarn('[worker/risk-control]', '打手频繁触发限流，进入二级风控惩罚锁定', {
|
||
primaryKey,
|
||
workerId,
|
||
ip,
|
||
violationCount: tracker.violationCount,
|
||
penaltyMinutes,
|
||
})
|
||
|
||
// 显著事件留痕（低频；被拒请求本身不落库）
|
||
recordWorkerRiskEvent({
|
||
eventType: 'tier2_penalized',
|
||
targetKey: primaryKey,
|
||
workerId,
|
||
ip,
|
||
violationCount: tracker.violationCount,
|
||
windowCount: tracker.windowCount,
|
||
penaltyUntil: new Date(tracker.penalizedUntil).toISOString(),
|
||
reason: `违规 ${tracker.violationCount} 次，进入 ${penaltyMinutes} 分钟静默惩罚期`,
|
||
detail: { penaltyMinutes, config },
|
||
})
|
||
|
||
const error = createHttpError(
|
||
`检测到连续异常高频请求，已触发 ${penaltyMinutes} 分钟安全风控限制，请停止刷单脚本`,
|
||
{
|
||
statusCode: 429,
|
||
errorCode: 'risk_penalized',
|
||
},
|
||
)
|
||
res.status(429).json(buildErrorPayload(error, '触发安全风控限制'))
|
||
return
|
||
}
|
||
|
||
// 仅 Tier 1 阶段超限，返回轻量 429
|
||
res.setHeader('Retry-After', '2')
|
||
logWarn('[worker/risk-control]', '大厅请求超过正常频率', {
|
||
primaryKey,
|
||
workerId,
|
||
ip,
|
||
windowCount: tracker.windowCount,
|
||
violationCount: tracker.violationCount,
|
||
})
|
||
|
||
const error = createHttpError('抢单大厅刷新过于频繁，请稍后再试', {
|
||
statusCode: 429,
|
||
errorCode: 'hall_rate_limited',
|
||
})
|
||
res.status(429).json(buildErrorPayload(error, '大厅请求过于频繁'))
|
||
}
|
||
}
|
||
|
||
/** 触发 Tier 3 自动封禁 */
|
||
async function executeAutoFreezeWorker(workerId: number, ip: string, violationCount: number) {
|
||
try {
|
||
const now = nowIso()
|
||
const reason = `系统风控：检测到高频自动化刷接口（违规超 ${violationCount} 次），多次警告后未停止，已被系统自动封禁`
|
||
const updated = await updateWorkerUser(workerId, {
|
||
status: 'frozen',
|
||
frozen_reason: reason,
|
||
frozen_at: now,
|
||
updated_at: now,
|
||
})
|
||
// updateWorkerUser 在账号不存在时返回 null 而**不抛错**。必须显式判断，
|
||
// 否则会为一次根本没生效的封禁写下"已永久封禁"的日志与留痕，误导事后追溯。
|
||
if (!updated) {
|
||
logError('[worker/risk-control]', '自动封禁未生效：账号不存在，已跳过留痕', {
|
||
workerId,
|
||
ip,
|
||
violationCount,
|
||
})
|
||
return
|
||
}
|
||
// 强制踢掉该打手所有现存登录设备
|
||
await revokeAllWorkerSessions(workerId, now)
|
||
logError('[worker/risk-control]', '打手账号已被系统自动永久封禁！', {
|
||
workerId,
|
||
ip,
|
||
violationCount,
|
||
reason,
|
||
})
|
||
|
||
recordWorkerRiskEvent({
|
||
eventType: 'tier3_autofrozen',
|
||
targetKey: `worker:${workerId}`,
|
||
workerId,
|
||
ip,
|
||
violationCount,
|
||
reason,
|
||
})
|
||
} catch (err) {
|
||
logError('[worker/risk-control]', '自动封禁打手账号执行异常', { workerId, ip, err })
|
||
}
|
||
}
|
||
|
||
/**
|
||
* 风控事件留痕。**fire-and-forget**：留痕失败绝不能影响请求结果，
|
||
* 更不允许因为一次数据库抖动把大厅请求变成 500。
|
||
*/
|
||
function recordWorkerRiskEvent(input: Parameters<typeof insertWorkerRiskEvent>[0]): void {
|
||
void insertWorkerRiskEvent(input).catch((err) => {
|
||
logError('[worker/risk-control]', '风控事件留痕失败', { err, eventType: input.eventType })
|
||
})
|
||
}
|
||
|
||
export type WorkerRiskTrackerSnapshot = {
|
||
key: string
|
||
workerId: number | null
|
||
ip: string
|
||
banned: boolean
|
||
penalizedUntil: number
|
||
violationCount: number
|
||
windowCount: number
|
||
rejectedCount: number
|
||
lastSeenAt: number
|
||
}
|
||
|
||
/**
|
||
* 后台展示用：当前仍在生效（封禁 / 惩罚中 / 本轮有违规）的内存风控项。
|
||
* 这些数据不落库，进程重启即清零——后台需如实标注，避免被误读为历史累计。
|
||
*/
|
||
export function listActiveWorkerRiskTrackers(): WorkerRiskTrackerSnapshot[] {
|
||
const now = Date.now()
|
||
const items: WorkerRiskTrackerSnapshot[] = []
|
||
|
||
for (const [key, tracker] of riskTrackers.entries()) {
|
||
if (!tracker.banned && tracker.penalizedUntil <= now && tracker.violationCount <= 0) {
|
||
continue
|
||
}
|
||
items.push({
|
||
key,
|
||
workerId: tracker.lastWorkerId,
|
||
ip: tracker.lastIp,
|
||
banned: tracker.banned,
|
||
penalizedUntil: tracker.penalizedUntil,
|
||
violationCount: tracker.violationCount,
|
||
windowCount: tracker.windowCount,
|
||
rejectedCount: tracker.rejectedCount,
|
||
lastSeenAt: tracker.lastSeenAt,
|
||
})
|
||
}
|
||
|
||
return items.sort((a, b) => b.lastSeenAt - a.lastSeenAt).slice(0, TRACKER_SNAPSHOT_LIMIT)
|
||
}
|
||
|
||
/**
|
||
* 解除某个打手（及可选 IP）的内存风控状态。
|
||
*
|
||
* 后台解封账号时**必须**连同调用：把数据库改回 active 只解决鉴权层，
|
||
* 内存里的 banned / 惩罚标记不会自己消失（30 秒对账虽能兜底恢复，但后台点击后应立即生效）。
|
||
*/
|
||
export function releaseWorkerRiskTracker(input: {
|
||
workerId?: number | null
|
||
ip?: string | null
|
||
}): void {
|
||
const workerId = Number(input.workerId)
|
||
if (Number.isInteger(workerId) && workerId > 0) {
|
||
riskTrackers.delete(`worker:${workerId}`)
|
||
}
|
||
const ip = String(input.ip || '').trim()
|
||
if (ip) {
|
||
riskTrackers.delete(`ip:${ip}`)
|
||
}
|
||
}
|
||
|
||
/**
|
||
* 后台人工封禁：同时置位内存标记，避免出现「数据库已 frozen、内存却仍放行」的反向不一致。
|
||
* 仅支持按账号封禁——按 IP 封禁没有可对账的事实来源，会永久生效，因此不开放。
|
||
*/
|
||
export function markWorkerRiskBanned(workerId: number): void {
|
||
const id = Number(workerId)
|
||
if (!Number.isInteger(id) || id <= 0) return
|
||
const now = Date.now()
|
||
const tracker = getOrCreateTracker(`worker:${id}`, now)
|
||
tracker.banned = true
|
||
tracker.bannedAt = now
|
||
tracker.bannedVerifiedAt = now
|
||
tracker.lastWorkerId = id
|
||
tracker.lastSeenAt = now
|
||
}
|
||
|
||
/** 测试辅助：清空风控记录 */
|
||
export function resetWorkerRiskTrackersForTest(): void {
|
||
riskTrackers.clear()
|
||
lastCleanupAt = 0
|
||
}
|
||
|
||
/** 获取内部 tracker 状态供测试断言 */
|
||
export function getWorkerRiskTrackerForTest(key: string): WorkerRiskTracker | undefined {
|
||
return riskTrackers.get(key)
|
||
}
|
||
|
||
function cleanupExpiredTrackers(now: number): void {
|
||
if (now - lastCleanupAt < 60_000) return
|
||
lastCleanupAt = now
|
||
for (const [key, tracker] of riskTrackers.entries()) {
|
||
if (
|
||
!tracker.banned &&
|
||
tracker.penalizedUntil <= now &&
|
||
tracker.windowResetAt <= now &&
|
||
tracker.violationResetAt <= now
|
||
) {
|
||
riskTrackers.delete(key)
|
||
}
|
||
}
|
||
// 硬上限防泄漏
|
||
if (riskTrackers.size > 20_000) {
|
||
const excess = riskTrackers.size - 20_000
|
||
let count = 0
|
||
for (const key of riskTrackers.keys()) {
|
||
riskTrackers.delete(key)
|
||
count++
|
||
if (count >= excess) break
|
||
}
|
||
}
|
||
}
|