fix: 标准化钉钉限流重试策略
This commit is contained in:
WuYongJie
parent
5be0b308bc
commit
3fb60314a2
@ -29,4 +29,6 @@ YYD_NOTIFICATION_PROJECTS_JSON=[{"code":"material_library","name":"素材库","s
|
||||
|
||||
WORKER_ENABLED=true
|
||||
WORKER_INTERVAL_MS=1000
|
||||
# 钉钉发送请求的最小启动间隔，统一限制在 100~5000ms；默认 250ms（最多约 4 条/秒）。
|
||||
WORKER_SEND_MIN_INTERVAL_MS=250
|
||||
READ_STATUS_INTERVAL_MS=5000
|
||||
|
||||
@ -170,4 +170,4 @@ POST /v1/auth/tickets/redeem
|
||||
- `YYD_NOTIFICATION_PUBLIC_BASE_URL`：通知中心公网 HTTPS 基地址，生产为反向代理后的 `/notification-center` 地址。
|
||||
- `DINGTALK_LOGIN_TICKET_TTL_SECONDS`：30~60 秒，默认 60 秒。
|
||||
|
||||
新建任务最多投递 5 次，明确的限流等未受理结果按 1、5、15、30 分钟退避。发送调用若在请求已发出后发生网络断开、超时、HTTP 408/5xx、钉钉成功响应缺少 `processQueryKey`，或进程在发送中途崩溃并遗留过期租约，会以 `DELIVERY_OUTCOME_UNKNOWN` 结束并停止自动重试，避免不确定结果造成重复单聊。历史任务保留创建时的最大尝试次数，不通过迁移篡改。
|
||||
新建任务最多投递 5 次，钉钉 HTTP 429、`QpsLimit` 和其他明确限流统一归一为 `FLOW_CONTROLLED`，按 1、5、15、30 分钟退避；接入方只观察统一状态和错误码，不解析钉钉厂商错误码。中心默认将发送请求启动间隔限制为至少 250ms（`WORKER_SEND_MIN_INTERVAL_MS` 可在 100～5000ms 内调整），避免整点任务集中冲击下游。发送调用若在请求已发出后发生网络断开、超时、非限流 HTTP 408/5xx、钉钉成功响应缺少 `processQueryKey`，或进程在发送中途崩溃并遗留过期租约，会以 `DELIVERY_OUTCOME_UNKNOWN` 结束并停止自动重试，避免不确定结果造成重复单聊。历史任务保留创建时的最大尝试次数，不通过迁移篡改。
|
||||
|
||||
@ -20,6 +20,7 @@ services:
|
||||
YYD_NOTIFICATION_ADMIN_API_SECRET: ${YYD_NOTIFICATION_ADMIN_API_SECRET}
|
||||
YYD_NOTIFICATION_CONFIG_ENCRYPTION_KEY: ${YYD_NOTIFICATION_CONFIG_ENCRYPTION_KEY}
|
||||
WORKER_ENABLED: ${WORKER_ENABLED:-true}
|
||||
WORKER_SEND_MIN_INTERVAL_MS: ${WORKER_SEND_MIN_INTERVAL_MS:-250}
|
||||
volumes:
|
||||
- notification_data:/app/data
|
||||
|
||||
|
||||
@ -26,7 +26,14 @@
|
||||
|
||||
业务项目一般不需要自己重发 `pending / sending / retrying / sent / unread` 状态。通知中心会处理明确可重试的失败并查询已读状态。
|
||||
|
||||
## 3. 常见错误码
|
||||
## 3. 标准边界
|
||||
|
||||
- 业务项目必须按统一请求契约提交 `recipientUserId` 或已登记的 `recipientAlias`、标题、正文、跳转地址和幂等键；通知中心不识别采购流程、供应商账号、邮箱等项目私有字段。
|
||||
- 收件人解析和业务人员配置由接入方负责。姓名、邮箱、外部联系人 ID 和自行拼接的标识不能代替钉钉内部员工 `userId`。
|
||||
- 接入方以 HTTP 状态和投递 `status` 作为自动控制流程依据；`lastErrorCode` 用于诊断和人工处置，不得根据未知错误码自行重发。
|
||||
- 本规范公布的统一 `lastErrorCode` 语义稳定；尚未归一化的钉钉厂商错误码或附在 `lastErrorMessage` 中的厂商信息只供通知中心维护人排查，不构成对接契约。
|
||||
|
||||
## 4. 常见错误码
|
||||
|
||||
| 错误码 | 原因 | 处理方式 |
|
||||
| --- | --- | --- |
|
||||
@ -46,10 +53,14 @@
|
||||
| `IDEMPOTENCY_CONFLICT` | 同一幂等键对应了不同内容 | 修复幂等键或业务数据，不得随机换键规避 |
|
||||
| `NOT_FOUND` | 记录不存在或跨项目访问 | 检查投递 ID 和项目身份 |
|
||||
| `INTERNAL_ERROR` | 通知中心内部异常 | 原幂等键退避重试并通知维护人员 |
|
||||
| `FLOW_CONTROLLED` | 钉钉明确拒绝本次请求并要求限流 | 接入方等待，通知中心内部自动有限重试 |
|
||||
| `INVALID_USER` | 钉钉确认收件人内部员工 `userId` 无效或不存在 | 接入方修正人员数据，不得改用邮箱或姓名重试 |
|
||||
| `FILTERED_USER` | 收件人不在统一机器人可见范围内 | 由通知中心维护人核对应用权限范围 |
|
||||
| `DELIVERY_OUTCOME_UNKNOWN` | 请求可能已经送达，无法安全确认 | 停止自动重试，人工核验后决定是否补发 |
|
||||
|
||||
投递进入 `failed` 时，还应记录返回中的 `lastErrorCode` 和 `lastErrorMessage`，但不要直接把内部错误原文展示给最终用户。
|
||||
|
||||
## 4. 业务端重试原则
|
||||
## 5. 业务端重试原则
|
||||
|
||||
1. 连接失败或 HTTP `500` 时可以有限重试，必须复用原 `idempotencyKey`，每次生成新 nonce。
|
||||
2. HTTP `200 / 202` 后不要重新创建通知，改为查询返回的投递 ID。
|
||||
@ -57,7 +68,7 @@
|
||||
4. 通知中心内部会处理明确未受理的钉钉限流等情况，业务项目不得再叠加无限重试。
|
||||
5. `DELIVERY_OUTCOME_UNKNOWN` 表示发送结果无法确定。为避免用户收到重复单聊，通知中心会停止自动重试；业务负责人确认后才能人工补发。
|
||||
|
||||
## 5. 建议监控
|
||||
## 6. 建议监控
|
||||
|
||||
业务项目至少记录：
|
||||
|
||||
|
||||
@ -252,6 +252,7 @@ center_lines = [
|
||||
'YYD_NOTIFICATION_CONFIG_ENCRYPTION_KEY=' + quoted(config_encryption_key),
|
||||
'WORKER_ENABLED=true',
|
||||
'WORKER_INTERVAL_MS=1000',
|
||||
'WORKER_SEND_MIN_INTERVAL_MS=250',
|
||||
'READ_STATUS_INTERVAL_MS=5000',
|
||||
]
|
||||
target_path.write_text('\n'.join(center_lines) + '\n', encoding='utf-8')
|
||||
|
||||
@ -13,6 +13,7 @@ test('生产发布面透传免登配置并保留已有项目 login', () => {
|
||||
'DINGTALK_CORP_ID',
|
||||
'YYD_NOTIFICATION_PUBLIC_BASE_URL',
|
||||
'DINGTALK_LOGIN_TICKET_TTL_SECONDS',
|
||||
'WORKER_SEND_MIN_INTERVAL_MS',
|
||||
]) {
|
||||
assert.match(compose, new RegExp(key))
|
||||
assert.match(example, new RegExp(key))
|
||||
|
||||
@ -2,7 +2,12 @@ import assert from 'node:assert/strict'
|
||||
import test from 'node:test'
|
||||
import { DingTalkError, sendRobotMessage } from '../dingtalk'
|
||||
import prisma from '../prisma'
|
||||
import { classifyDeliveryFailure, cleanupDeliveries, quarantineExpiredSendingLeases } from '../worker'
|
||||
import {
|
||||
classifyDeliveryFailure,
|
||||
cleanupDeliveries,
|
||||
normalizeWorkerSendMinIntervalMs,
|
||||
quarantineExpiredSendingLeases,
|
||||
} from '../worker'
|
||||
|
||||
test('token 失效先刷新重试一次，避免缓存过期导致批量误判终态', async () => {
|
||||
const originalFetch = global.fetch
|
||||
@ -192,6 +197,96 @@ test('机器人 ActionCard 原样发送统一标题、分段正文和按钮跳
|
||||
}
|
||||
})
|
||||
|
||||
test('钉钉 HTTP 403 QPS 限流归一为中心标准错误并进入自动重试', async () => {
|
||||
const originalFetch = global.fetch
|
||||
const oldAppKey = process.env.DINGTALK_APP_KEY
|
||||
const oldAppSecret = process.env.DINGTALK_APP_SECRET
|
||||
const oldRobotCode = process.env.DINGTALK_ROBOT_CODE
|
||||
process.env.DINGTALK_APP_KEY = 'test-app-key'
|
||||
process.env.DINGTALK_APP_SECRET = 'test-app-secret'
|
||||
process.env.DINGTALK_ROBOT_CODE = 'test-robot-code'
|
||||
global.fetch = (async (input) => {
|
||||
if (String(input).includes('/gettoken?')) {
|
||||
return new Response(JSON.stringify({ errcode: 0, access_token: 'qps-test-token', expires_in: 7200 }), { status: 200 })
|
||||
}
|
||||
return new Response(JSON.stringify({
|
||||
code: 'Forbidden.AccessDenied.QpsLimitForApi',
|
||||
message: '当前所有钉钉应用调用该接口次数过多，超出了该接口承受的最大qps，请求被暂时限制了',
|
||||
}), { status: 403 })
|
||||
}) as typeof fetch
|
||||
|
||||
try {
|
||||
let caught: unknown
|
||||
try {
|
||||
await sendRobotMessage({ recipientUserId: 'ding-user-1', title: '测试', content: '测试', actionTitle: '查看', actionUrl: 'https://example.com' })
|
||||
} catch (error) {
|
||||
caught = error
|
||||
}
|
||||
assert.ok(caught instanceof DingTalkError)
|
||||
assert.equal(caught.code, 'FLOW_CONTROLLED')
|
||||
assert.equal(caught.providerCode, 'Forbidden.AccessDenied.QpsLimitForApi')
|
||||
assert.equal(caught.deterministic, false)
|
||||
assert.equal(caught.outcomeUnknown, false)
|
||||
assert.deepEqual(classifyDeliveryFailure({ error: caught, attemptCount: 1, maxAttempts: 5 }), {
|
||||
failed: false,
|
||||
errorCode: 'FLOW_CONTROLLED',
|
||||
errorMessage: '[Forbidden.AccessDenied.QpsLimitForApi] 当前所有钉钉应用调用该接口次数过多，超出了该接口承受的最大qps，请求被暂时限制了',
|
||||
})
|
||||
} finally {
|
||||
global.fetch = originalFetch
|
||||
process.env.DINGTALK_APP_KEY = oldAppKey
|
||||
process.env.DINGTALK_APP_SECRET = oldAppSecret
|
||||
process.env.DINGTALK_ROBOT_CODE = oldRobotCode
|
||||
}
|
||||
})
|
||||
|
||||
test('钉钉 staff 不存在归一为收件人无效并保持终态失败', async () => {
|
||||
const originalFetch = global.fetch
|
||||
const oldAppKey = process.env.DINGTALK_APP_KEY
|
||||
const oldAppSecret = process.env.DINGTALK_APP_SECRET
|
||||
const oldRobotCode = process.env.DINGTALK_ROBOT_CODE
|
||||
process.env.DINGTALK_APP_KEY = 'test-app-key'
|
||||
process.env.DINGTALK_APP_SECRET = 'test-app-secret'
|
||||
process.env.DINGTALK_ROBOT_CODE = 'test-robot-code'
|
||||
global.fetch = (async (input) => {
|
||||
if (String(input).includes('/gettoken?')) {
|
||||
return new Response(JSON.stringify({ errcode: 0, access_token: 'invalid-user-test-token', expires_in: 7200 }), { status: 200 })
|
||||
}
|
||||
return new Response(JSON.stringify({
|
||||
code: 'staffId.notExisted',
|
||||
message: '错误描述: staff 不存在；解决方案:请确认 userIds 是否正确；',
|
||||
}), { status: 400 })
|
||||
}) as typeof fetch
|
||||
|
||||
try {
|
||||
let caught: unknown
|
||||
try {
|
||||
await sendRobotMessage({ recipientUserId: 'ding-user-invalid', title: '测试', content: '测试', actionTitle: '查看', actionUrl: 'https://example.com' })
|
||||
} catch (error) {
|
||||
caught = error
|
||||
}
|
||||
assert.ok(caught instanceof DingTalkError)
|
||||
assert.equal(caught.code, 'INVALID_USER')
|
||||
assert.equal(caught.providerCode, 'staffId.notExisted')
|
||||
assert.equal(caught.deterministic, true)
|
||||
assert.equal(classifyDeliveryFailure({ error: caught, attemptCount: 1, maxAttempts: 5 }).failed, true)
|
||||
} finally {
|
||||
global.fetch = originalFetch
|
||||
process.env.DINGTALK_APP_KEY = oldAppKey
|
||||
process.env.DINGTALK_APP_SECRET = oldAppSecret
|
||||
process.env.DINGTALK_ROBOT_CODE = oldRobotCode
|
||||
}
|
||||
})
|
||||
|
||||
test('中心发送间隔使用安全默认值并限制配置边界', () => {
|
||||
assert.equal(normalizeWorkerSendMinIntervalMs(undefined), 250)
|
||||
assert.equal(normalizeWorkerSendMinIntervalMs(''), 250)
|
||||
assert.equal(normalizeWorkerSendMinIntervalMs('invalid'), 250)
|
||||
assert.equal(normalizeWorkerSendMinIntervalMs('0'), 100)
|
||||
assert.equal(normalizeWorkerSendMinIntervalMs('375'), 375)
|
||||
assert.equal(normalizeWorkerSendMinIntervalMs('99999'), 5000)
|
||||
})
|
||||
|
||||
test('明确限流可安全重试，确定性错误和耗尽次数则停止', () => {
|
||||
assert.equal(classifyDeliveryFailure({
|
||||
error: new DingTalkError('限流', 'FLOW_CONTROLLED'), attemptCount: 1, maxAttempts: 4,
|
||||
|
||||
@ -8,6 +8,7 @@ export class DingTalkError extends Error {
|
||||
public readonly code: string,
|
||||
public readonly deterministic = false,
|
||||
public readonly outcomeUnknown = false,
|
||||
public readonly providerCode?: string,
|
||||
) {
|
||||
super(message)
|
||||
this.name = 'DingTalkError'
|
||||
@ -32,18 +33,34 @@ const requestJson = async (
|
||||
})
|
||||
const data = await response.json().catch(() => ({})) as DingTalkPayload
|
||||
if (!response.ok) {
|
||||
const code = clean(data.code) || `HTTP_${response.status}`
|
||||
const providerCode = clean(data.code) || `HTTP_${response.status}`
|
||||
const message = clean(data.message) || `钉钉接口返回 HTTP ${response.status}`
|
||||
const outcomeUnknown = Boolean(options?.deliveryOutcomeCanBeUnknown && (response.status === 408 || response.status >= 500))
|
||||
const code = normalizeDingTalkErrorCode(providerCode, message, response.status)
|
||||
const rateLimited = code === 'FLOW_CONTROLLED'
|
||||
const outcomeUnknown = Boolean(
|
||||
options?.deliveryOutcomeCanBeUnknown &&
|
||||
!rateLimited &&
|
||||
(response.status === 408 || response.status >= 500),
|
||||
)
|
||||
throw new DingTalkError(
|
||||
message,
|
||||
code,
|
||||
response.status >= 400 && response.status < 500 && response.status !== 408 && response.status !== 429,
|
||||
!rateLimited && response.status >= 400 && response.status < 500 && response.status !== 408 && response.status !== 429,
|
||||
outcomeUnknown,
|
||||
providerCode !== code ? providerCode : undefined,
|
||||
)
|
||||
}
|
||||
if (data.errcode !== undefined && Number(data.errcode) !== 0) {
|
||||
throw new DingTalkError(clean(data.errmsg) || '钉钉接口调用失败', String(data.errcode), isDeterministic(data.errcode, data.errmsg))
|
||||
const providerCode = String(data.errcode)
|
||||
const message = clean(data.errmsg) || '钉钉接口调用失败'
|
||||
const code = normalizeDingTalkErrorCode(providerCode, message)
|
||||
throw new DingTalkError(
|
||||
message,
|
||||
code,
|
||||
isDeterministic(code, message),
|
||||
false,
|
||||
providerCode !== code ? providerCode : undefined,
|
||||
)
|
||||
}
|
||||
return data
|
||||
} catch (error: any) {
|
||||
@ -64,9 +81,27 @@ const requestJson = async (
|
||||
|
||||
const clean = (value: unknown) => String(value || '').trim()
|
||||
const stringList = (value: unknown) => Array.isArray(value) ? value.map(clean).filter(Boolean) : []
|
||||
const isDeterministic = (code: unknown, message: unknown) =>
|
||||
['40014', '40056', '40057', '40078', '41001', '41009', '60111', '60121', '70020'].includes(String(code)) ||
|
||||
|
||||
const isDingTalkRateLimitError = (code: unknown, message: unknown, status?: number) => {
|
||||
const normalizedCode = clean(code).toLowerCase()
|
||||
const normalizedMessage = clean(message)
|
||||
return status === 429 ||
|
||||
/qps.?limit|rate.?limit|too.?many.?requests|throttl/i.test(normalizedCode) ||
|
||||
/最大\s*qps|qps\s*限制|请求.*(?:过多|限流|暂时限制)|too many requests|rate limit/i.test(normalizedMessage)
|
||||
}
|
||||
|
||||
const normalizeDingTalkErrorCode = (providerCode: unknown, message: unknown, status?: number) => {
|
||||
const code = clean(providerCode)
|
||||
if (isDingTalkRateLimitError(code, message, status)) return 'FLOW_CONTROLLED'
|
||||
if (code.toLowerCase() === 'staffid.notexisted') return 'INVALID_USER'
|
||||
return code
|
||||
}
|
||||
|
||||
const isDeterministic = (code: unknown, message: unknown) => {
|
||||
if (clean(code) === 'FLOW_CONTROLLED') return false
|
||||
return ['40014', '40056', '40057', '40078', '41001', '41009', '60111', '60121', '70020', 'INVALID_USER'].includes(String(code)) ||
|
||||
/不存在|不在可见范围|无权限|invalid user|forbidden user|userid/i.test(clean(message))
|
||||
}
|
||||
|
||||
// token 过期属于可恢复错误：先作废本地缓存、用新 token 重试一次；仍失败再按原语义上抛。
|
||||
const AUTH_EXPIRED_CODES = new Set(['40014', 'HTTP_401', 'InvalidAuthentication'])
|
||||
|
||||
@ -6,6 +6,9 @@ const workerId = `${os.hostname()}:${process.pid}:${Math.random().toString(36).s
|
||||
const backoffMinutes = [1, 5, 15, 30]
|
||||
// 单 tick 批量消费，避免积压时每秒只推进一条。
|
||||
const MAX_DELIVERIES_PER_TICK = 10
|
||||
const DEFAULT_SEND_MIN_INTERVAL_MS = 250
|
||||
const MIN_SEND_MIN_INTERVAL_MS = 100
|
||||
const MAX_SEND_MIN_INTERVAL_MS = 5000
|
||||
// sent/unread 超过该天数仍无终态则停止回查；终态记录保留该天数后清理。
|
||||
const READ_STATUS_EXPIRY_DAYS = 14
|
||||
const DELIVERY_RETENTION_DAYS = 90
|
||||
@ -14,6 +17,20 @@ let busy = false
|
||||
let readBusy = false
|
||||
const outcomeUnknownMessage = '钉钉发送结果不确定，请先在钉钉侧人工核验，系统不会自动重试'
|
||||
|
||||
export const normalizeWorkerSendMinIntervalMs = (value: unknown) => {
|
||||
if (value === undefined || value === null || String(value).trim() === '') return DEFAULT_SEND_MIN_INTERVAL_MS
|
||||
const parsed = Number(value)
|
||||
if (!Number.isFinite(parsed)) return DEFAULT_SEND_MIN_INTERVAL_MS
|
||||
return Math.min(MAX_SEND_MIN_INTERVAL_MS, Math.max(MIN_SEND_MIN_INTERVAL_MS, Math.round(parsed)))
|
||||
}
|
||||
|
||||
const deliveryErrorMessage = (error: unknown) => {
|
||||
if (error instanceof DingTalkError && error.providerCode) {
|
||||
return `[${error.providerCode}] ${error.message}`.slice(0, 2000)
|
||||
}
|
||||
return String((error as any)?.message || '通知发送失败').slice(0, 2000)
|
||||
}
|
||||
|
||||
export const quarantineExpiredSendingLeases = async (now = new Date()) => prisma.notificationDelivery.updateMany({
|
||||
where: {
|
||||
status: 'sending',
|
||||
@ -46,7 +63,7 @@ export const classifyDeliveryFailure = (input: {
|
||||
return {
|
||||
failed: deterministic || exhausted,
|
||||
errorCode: error instanceof DingTalkError ? error.code : 'UNKNOWN',
|
||||
errorMessage: String((error as any)?.message || '通知发送失败').slice(0, 2000),
|
||||
errorMessage: deliveryErrorMessage(error),
|
||||
}
|
||||
}
|
||||
|
||||
@ -72,10 +89,11 @@ const claimNext = async () => {
|
||||
return prisma.notificationDelivery.findUnique({ where: { id: candidate.id } })
|
||||
}
|
||||
|
||||
const processOne = async (): Promise<boolean> => {
|
||||
const processOne = async (beforeSend?: () => Promise<void>): Promise<boolean> => {
|
||||
const delivery = await claimNext()
|
||||
if (!delivery) return false
|
||||
try {
|
||||
await beforeSend?.()
|
||||
const processQueryKey = await sendRobotMessage({
|
||||
recipientUserId: delivery.recipientUserId,
|
||||
title: delivery.renderedTitle,
|
||||
@ -173,12 +191,19 @@ export const startWorker = () => {
|
||||
if (process.env.WORKER_ENABLED === 'false') return
|
||||
const interval = Math.max(500, Number(process.env.WORKER_INTERVAL_MS) || 1000)
|
||||
const readInterval = Math.max(2000, Number(process.env.READ_STATUS_INTERVAL_MS) || 5000)
|
||||
const sendMinInterval = normalizeWorkerSendMinIntervalMs(process.env.WORKER_SEND_MIN_INTERVAL_MS)
|
||||
let lastSendStartedAt = 0
|
||||
const waitForSendSlot = async () => {
|
||||
const waitMs = Math.max(0, sendMinInterval - (Date.now() - lastSendStartedAt))
|
||||
if (waitMs > 0) await new Promise((resolve) => setTimeout(resolve, waitMs))
|
||||
lastSendStartedAt = Date.now()
|
||||
}
|
||||
const tick = async () => {
|
||||
if (busy) return
|
||||
busy = true
|
||||
try {
|
||||
for (let processed = 0; processed < MAX_DELIVERIES_PER_TICK; processed++) {
|
||||
const claimed = await processOne()
|
||||
const claimed = await processOne(waitForSendSlot)
|
||||
if (!claimed) break
|
||||
}
|
||||
} finally {
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user