降低云触手 bind_info 限频(110001)命中率

识别中文过于频繁错误并指数退避;bind_info 加长间隔与短缓存;
领取轮询放慢并去掉同轮重复 probe,减轻上游压力。
This commit is contained in:
yml2213
2026-07-10 21:07:19 +08:00
parent 5ed6890cc5
commit 50d32be0ea
5 changed files with 221 additions and 41 deletions
@@ -1,7 +1,6 @@
import { import {
normalizeKuaishouCloudFlow, normalizeKuaishouCloudFlow,
prepareKuaishouCloudFulfillmentTask, prepareKuaishouCloudFulfillmentTask,
probeKuaishouCloudTaskBindUrl,
refreshKuaishouCloudTaskRoleInfo, refreshKuaishouCloudTaskRoleInfo,
} from '../fulfillment/kuaishou-cloud/index.js' } from '../fulfillment/kuaishou-cloud/index.js'
import type { TaskRow } from '../../types/repository/rows.js' import type { TaskRow } from '../../types/repository/rows.js'
@@ -27,18 +26,13 @@ export async function syncKuaishouCloudRoleInfo(task: TaskRow) {
} }
try { try {
const probed = await probeKuaishouCloudTaskBindUrl(task, { // 只走 refresh:内部会按间隔 probe 绑链,必要时再查 bind_info
source: 'claim_page_polling_bind_url_probe', // 不再 forceProbe + 前置 probe,避免同轮重复打上游
actor: { source: 'system' },
})
task = probed.task || task
const result = await refreshKuaishouCloudTaskRoleInfo(task, { const result = await refreshKuaishouCloudTaskRoleInfo(task, {
source: 'claim_page_polling', source: 'claim_page_polling',
actor: { source: 'system' }, actor: { source: 'system' },
recordEvent: false, recordEvent: false,
// 轮询阶段始终允许探测绑链,默认角色仅作诊断 forceProbe: false,
forceProbe: true,
}) })
return result.task || task return result.task || task
} catch { } catch {
@@ -2,13 +2,23 @@ import test from 'node:test'
import assert from 'node:assert/strict' import assert from 'node:assert/strict'
import http from 'node:http' import http from 'node:http'
import { cloudtentaclesRequest } from './http-client.js' import { cloudtentaclesRequest, isHighFrequencyMessage } from './http-client.js'
type MockResponse = { type MockResponse = {
status?: number status?: number
body?: unknown body?: unknown
} }
test('isHighFrequencyMessage recognizes chinese 110001 and english high-frequency', () => {
assert.equal(
isHighFrequencyMessage('未知错误码: (110001) 抱歉,请求过于频繁,请稍后再试!谢谢!', -1),
true,
)
assert.equal(isHighFrequencyMessage('high-frequency Request'), true)
assert.equal(isHighFrequencyMessage('ok', 0), false)
assert.equal(isHighFrequencyMessage('其他错误', 1), false)
})
test('cloudtentaclesRequest retries high-frequency business errors', async () => { test('cloudtentaclesRequest retries high-frequency business errors', async () => {
let callCount = 0 let callCount = 0
const server = await createMockServer(() => { const server = await createMockServer(() => {
@@ -37,6 +47,38 @@ test('cloudtentaclesRequest retries high-frequency business errors', async () =>
} }
}) })
test('cloudtentaclesRequest retries chinese 110001 rate-limit messages', async () => {
let callCount = 0
const server = await createMockServer(() => {
callCount += 1
return {
body:
callCount === 1
? {
code: -1,
message: '未知错误码: (110001) 抱歉,请求过于频繁,请稍后再试!谢谢!',
}
: { code: 0, data: 'ok' },
}
})
try {
const result = await cloudtentaclesRequest('/vn/bind_info', {
baseUrl: server.baseUrl,
token: 'cn-retry-token',
sourceKey: 'cn-retry-source',
rateLimitIntervalMs: 0,
rateLimitRetries: 1,
rateLimitRetryDelayMs: 1,
})
assert.equal(callCount, 2)
assert.equal(result.payload?.data, 'ok')
} finally {
await server.close()
}
})
test('cloudtentaclesRequest maps high-frequency exhaustion to 429', async () => { test('cloudtentaclesRequest maps high-frequency exhaustion to 429', async () => {
const server = await createMockServer(() => ({ const server = await createMockServer(() => ({
body: { code: 1, message: 'high-frequency Request' }, body: { code: 1, message: 'high-frequency Request' },
@@ -17,10 +17,13 @@ type NodeHttpResponse = {
bodyText: string bodyText: string
} }
const DEFAULT_RATE_LIMIT_INTERVAL_MS = 300 /** 通用接口默认最小间隔;bind_info 等敏感接口由调用方覆盖 */
const DEFAULT_RATE_LIMIT_INTERVAL_MS = 500
const DEFAULT_RATE_LIMIT_RETRIES = 2 const DEFAULT_RATE_LIMIT_RETRIES = 2
const DEFAULT_RATE_LIMIT_RETRY_DELAY_MS = 800 const DEFAULT_RATE_LIMIT_RETRY_DELAY_MS = 1_200
const MAX_RETRY_JITTER_MS = 180 const MAX_RETRY_JITTER_MS = 400
/** 命中上游限频后,额外冷却桶时间 */
const RATE_LIMIT_COOLDOWN_MS = 3_000
const bucketNextAvailableAt = new Map<string, number>() const bucketNextAvailableAt = new Map<string, number>()
let rateLimitQueue = Promise.resolve() let rateLimitQueue = Promise.resolve()
@@ -42,10 +45,18 @@ export async function cloudtentaclesRequest(pathname: unknown, options: JsonObje
token: options.token, token: options.token,
sourceKey: options.sourceKey, sourceKey: options.sourceKey,
deviceId: options.deviceId ?? config.deviceId, deviceId: options.deviceId ?? config.deviceId,
pathname: normalizedPathname,
isolatePath: options.rateLimitIsolatePath === true,
}) })
const rateLimitIntervalMs = normalizeNonNegativeInteger(options.rateLimitIntervalMs, DEFAULT_RATE_LIMIT_INTERVAL_MS) const rateLimitIntervalMs = normalizeNonNegativeInteger(
options.rateLimitIntervalMs,
resolveDefaultRateLimitIntervalMs(normalizedPathname),
)
const maxRetries = normalizeNonNegativeInteger(options.rateLimitRetries, DEFAULT_RATE_LIMIT_RETRIES) const maxRetries = normalizeNonNegativeInteger(options.rateLimitRetries, DEFAULT_RATE_LIMIT_RETRIES)
const retryDelayMs = normalizeNonNegativeInteger(options.rateLimitRetryDelayMs, DEFAULT_RATE_LIMIT_RETRY_DELAY_MS) const retryDelayMs = normalizeNonNegativeInteger(
options.rateLimitRetryDelayMs,
resolveDefaultRateLimitRetryDelayMs(normalizedPathname),
)
for (let attempt = 0; attempt <= maxRetries; attempt += 1) { for (let attempt = 0; attempt <= maxRetries; attempt += 1) {
try { try {
@@ -63,7 +74,14 @@ export async function cloudtentaclesRequest(pathname: unknown, options: JsonObje
if (!response.ok) { if (!response.ok) {
if (isCloudtentaclesHttpRateLimited(response.status) && attempt < maxRetries) { if (isCloudtentaclesHttpRateLimited(response.status) && attempt < maxRetries) {
await waitBeforeRateLimitRetry(attempt, normalizedPathname, method, response.status, retryDelayMs) await waitBeforeRateLimitRetry({
attempt,
pathname: normalizedPathname,
method,
status: response.status,
retryDelayMs,
bucketKey: rateLimitBucketKey,
})
continue continue
} }
@@ -98,10 +116,19 @@ export async function cloudtentaclesRequest(pathname: unknown, options: JsonObje
const message = String(payload?.message || payload?.msg || 'cloudtentacles 接口返回失败') const message = String(payload?.message || payload?.msg || 'cloudtentacles 接口返回失败')
const statusCode = Number(options.businessErrorStatusCode || 400) const statusCode = Number(options.businessErrorStatusCode || 400)
const errorCode = String(options.businessErrorCode || 'cloudtentacles_business_error') const errorCode = String(options.businessErrorCode || 'cloudtentacles_business_error')
const upstreamCode = payload?.code
if (isHighFrequencyMessage(message)) { if (isHighFrequencyMessage(message, upstreamCode)) {
if (attempt < maxRetries) { if (attempt < maxRetries) {
await waitBeforeRateLimitRetry(attempt, normalizedPathname, method, response.status, retryDelayMs) await waitBeforeRateLimitRetry({
attempt,
pathname: normalizedPathname,
method,
status: response.status,
retryDelayMs,
bucketKey: rateLimitBucketKey,
exponential: true,
})
continue continue
} }
@@ -111,6 +138,7 @@ export async function cloudtentaclesRequest(pathname: unknown, options: JsonObje
context: { context: {
pathname: normalizedPathname, pathname: normalizedPathname,
upstreamMessage: message, upstreamMessage: message,
upstreamCode,
}, },
}) })
} }
@@ -131,7 +159,7 @@ export async function cloudtentaclesRequest(pathname: unknown, options: JsonObje
status: response.status, status: response.status,
attempt: attempt + 1, attempt: attempt + 1,
errorCode, errorCode,
upstreamCode: payload?.code, upstreamCode,
upstreamMessage: message, upstreamMessage: message,
sourceKey: options.sourceKey, sourceKey: options.sourceKey,
accountLabel: options.accountLabel, accountLabel: options.accountLabel,
@@ -146,7 +174,7 @@ export async function cloudtentaclesRequest(pathname: unknown, options: JsonObje
method, method,
pathname: normalizedPathname, pathname: normalizedPathname,
upstreamStatus: response.status, upstreamStatus: response.status,
upstreamCode: payload?.code, upstreamCode,
upstreamMessage: message, upstreamMessage: message,
upstreamPayload: payload, upstreamPayload: payload,
requestContext: options.context, requestContext: options.context,
@@ -186,19 +214,60 @@ export async function cloudtentaclesRequest(pathname: unknown, options: JsonObje
}) })
} }
/** 导出供单测:识别上游限频文案/错误码(含中文 110001) */
export function isHighFrequencyMessage(message: unknown, code: unknown = null) {
const text = String(message || '').trim().toLowerCase()
const codeText = String(code ?? '').trim()
if (codeText === '110001' || Number(code) === 110001) {
return true
}
if (!text) {
return false
}
return (
text.includes('high-frequency request') ||
text.includes('high frequency') ||
text.includes('110001') ||
text.includes('过于频繁') ||
text.includes('请求过于频繁') ||
text.includes('too frequent') ||
text.includes('rate limit') ||
text.includes('too many requests')
)
}
function isCloudtentaclesExpiredMessage(message: unknown) { function isCloudtentaclesExpiredMessage(message: unknown) {
const normalized = String(message || '').trim().toLowerCase() const normalized = String(message || '').trim().toLowerCase()
return normalized.includes('expired') || normalized.includes('过期') || normalized.includes('失效') return normalized.includes('expired') || normalized.includes('过期') || normalized.includes('失效')
} }
function isHighFrequencyMessage(message: unknown) {
return String(message || '').trim().toLowerCase().includes('high-frequency request')
}
function isCloudtentaclesHttpRateLimited(status: unknown) { function isCloudtentaclesHttpRateLimited(status: unknown) {
return Number(status) === 429 return Number(status) === 429
} }
function resolveDefaultRateLimitIntervalMs(pathname: string) {
if (isBindInfoPath(pathname)) {
// bind_info 是 110001 高发接口,单独加长间隔
return 1_500
}
return DEFAULT_RATE_LIMIT_INTERVAL_MS
}
function resolveDefaultRateLimitRetryDelayMs(pathname: string) {
if (isBindInfoPath(pathname)) {
return 2_500
}
return DEFAULT_RATE_LIMIT_RETRY_DELAY_MS
}
function isBindInfoPath(pathname: string) {
const normalized = String(pathname || '').trim().toLowerCase()
return normalized.includes('bind_info') || normalized.endsWith('/vn/bind_info')
}
function truncateText(value: unknown, maxLength = 1000) { function truncateText(value: unknown, maxLength = 1000) {
const text = String(value || '') const text = String(value || '')
if (text.length <= maxLength) { if (text.length <= maxLength) {
@@ -264,14 +333,26 @@ async function waitForCloudtentaclesSlot(bucketKey: string, intervalMs: number)
} }
} }
async function waitBeforeRateLimitRetry( async function waitBeforeRateLimitRetry(options: {
attempt: number, attempt: number
pathname: string, pathname: string
method: string, method: string
status: number, status: number
retryDelayMs: number, retryDelayMs: number
) { bucketKey: string
const delayMs = retryDelayMs * (attempt + 1) + Math.floor(Math.random() * MAX_RETRY_JITTER_MS) exponential?: boolean
}) {
const { attempt, pathname, method, status, retryDelayMs, bucketKey, exponential } = options
const factor = exponential ? 2 ** attempt : attempt + 1
const delayMs = retryDelayMs * factor + Math.floor(Math.random() * MAX_RETRY_JITTER_MS)
// 命中限频后额外冷却账号桶,避免紧接着再打
const coolUntil = Date.now() + Math.max(delayMs, RATE_LIMIT_COOLDOWN_MS)
const current = bucketNextAvailableAt.get(bucketKey) || 0
if (coolUntil > current) {
bucketNextAvailableAt.set(bucketKey, coolUntil)
}
logWarn('[cloudtentacles/http]', '请求触发云触手限频,准备重试', { logWarn('[cloudtentacles/http]', '请求触发云触手限频,准备重试', {
method, method,
pathname, pathname,
@@ -287,17 +368,25 @@ function resolveRateLimitBucketKey({
token, token,
sourceKey, sourceKey,
deviceId, deviceId,
pathname,
isolatePath,
}: { }: {
baseUrl: unknown baseUrl: unknown
token: unknown token: unknown
sourceKey: unknown sourceKey: unknown
deviceId: unknown deviceId: unknown
pathname?: unknown
isolatePath?: boolean
}) { }) {
return [ const parts = [
String(baseUrl || '').trim(), String(baseUrl || '').trim(),
String(sourceKey || '').trim() || String(token || '').trim() || 'anonymous', String(sourceKey || '').trim() || String(token || '').trim() || 'anonymous',
String(deviceId || '').trim(), String(deviceId || '').trim(),
].join('|') ]
if (isolatePath) {
parts.push(String(pathname || '').trim())
}
return parts.join('|')
} }
function normalizeNonNegativeInteger(value: unknown, fallback: number) { function normalizeNonNegativeInteger(value: unknown, fallback: number) {
@@ -8,6 +8,21 @@ import { resolveCloudtentaclesConfig } from './helpers.js'
const BIND_URL_PROBE_MAX_BODY_BYTES = 64 * 1024 const BIND_URL_PROBE_MAX_BODY_BYTES = 64 * 1024
const AMS_SIGNATURE_EXPIRED_CODE = '99998' const AMS_SIGNATURE_EXPIRED_CODE = '99998'
/** 同一 vn 短时复用 bind_info,降低领取轮询撞 110001 概率 */
const BIND_INFO_CACHE_TTL_MS = 15_000
const BIND_INFO_RATE_LIMIT_INTERVAL_MS = 1_500
const BIND_INFO_RATE_LIMIT_RETRY_DELAY_MS = 2_500
type BindInfoResult = {
baseUrl: string
key: string
id: number
bindInfo: unknown
rawItem: JsonObject
raw: unknown
}
const bindInfoCache = new Map<string, { expiresAt: number; value: BindInfoResult }>()
type HeaderAdapter = { type HeaderAdapter = {
get(name: unknown): string | null get(name: unknown): string | null
@@ -258,6 +273,17 @@ export async function getCloudtentaclesBindInfo(payload: JsonObject = {}) {
const key = requireKey(payload.key, 'cloudtentacles 获取绑定信息缺少 key', 'cloudtentacles_vn_bind_info_missing_key') const key = requireKey(payload.key, 'cloudtentacles 获取绑定信息缺少 key', 'cloudtentacles_vn_bind_info_missing_key')
const id = requireId(payload.id, 'cloudtentacles 获取绑定信息缺少 id', 'cloudtentacles_vn_bind_info_missing_id') const id = requireId(payload.id, 'cloudtentacles 获取绑定信息缺少 id', 'cloudtentacles_vn_bind_info_missing_id')
const config = resolveCloudtentaclesConfig(payload) const config = resolveCloudtentaclesConfig(payload)
const sourceKey = String(payload.sourceKey || '').trim()
const cacheKey = `${sourceKey || token}|${id}|${key}`
const skipCache = payload.skipCache === true
const now = Date.now()
if (!skipCache) {
const cached = bindInfoCache.get(cacheKey)
if (cached && cached.expiresAt > now) {
return cached.value
}
}
const result = await cloudtentaclesRequest(config.vnBindInfoPath, { const result = await cloudtentaclesRequest(config.vnBindInfoPath, {
...config, ...config,
@@ -266,13 +292,18 @@ export async function getCloudtentaclesBindInfo(payload: JsonObject = {}) {
body: [{ id, key }], body: [{ id, key }],
businessErrorStatusCode: 401, businessErrorStatusCode: 401,
businessErrorCode: 'cloudtentacles_vn_bind_info_failed', businessErrorCode: 'cloudtentacles_vn_bind_info_failed',
// bind_info 是 110001 高发接口:加长间隔、独立路径桶、更长退避
rateLimitIntervalMs: BIND_INFO_RATE_LIMIT_INTERVAL_MS,
rateLimitRetryDelayMs: BIND_INFO_RATE_LIMIT_RETRY_DELAY_MS,
rateLimitRetries: 2,
rateLimitIsolatePath: true,
}) })
const items = Array.isArray(result.payload?.data) ? result.payload.data : [] const items = Array.isArray(result.payload?.data) ? result.payload.data : []
const matchedItem = items.find((item: JsonObject) => Number(item?.id || 0) === id) || items[0] || {} const matchedItem = items.find((item: JsonObject) => Number(item?.id || 0) === id) || items[0] || {}
const bindInfo = parseBindInfo(matchedItem?.bind_info) const bindInfo = parseBindInfo(matchedItem?.bind_info)
return { const value: BindInfoResult = {
baseUrl: config.baseUrl, baseUrl: config.baseUrl,
key, key,
id, id,
@@ -280,6 +311,22 @@ export async function getCloudtentaclesBindInfo(payload: JsonObject = {}) {
rawItem: isPlainObject(matchedItem) ? matchedItem : {}, rawItem: isPlainObject(matchedItem) ? matchedItem : {},
raw: result.payload ?? null, raw: result.payload ?? null,
} }
bindInfoCache.set(cacheKey, {
expiresAt: now + BIND_INFO_CACHE_TTL_MS,
value,
})
// 简单淘汰过期项,避免 map 无限增长
if (bindInfoCache.size > 500) {
for (const [entryKey, entry] of bindInfoCache) {
if (entry.expiresAt <= now) {
bindInfoCache.delete(entryKey)
}
}
}
return value
} }
export async function backCloudtentaclesVirtualNumber(payload: JsonObject = {}) { export async function backCloudtentaclesVirtualNumber(payload: JsonObject = {}) {
+14 -6
View File
@@ -1,13 +1,16 @@
import { isClaimInactiveTaskStatus } from '@/domain/task-status' import { isClaimInactiveTaskStatus } from '@/domain/task-status'
import type { ClaimSnapshot } from './claim-snapshot' import type { ClaimSnapshot } from './claim-snapshot'
export const BINDING_PREPARE_POLL_MS = 5_000 /** 绑链准备中:略放缓,减少空转 */
export const ROLE_FAST_POLL_MS = 10_000 export const BINDING_PREPARE_POLL_MS = 8_000
export const ROLE_SLOW_POLL_MS = 30_000 /** 角色快轮询:原 10s → 25s,降低 bind_info 压力 */
export const ROLE_IDLE_POLL_MS = 60_000 export const ROLE_FAST_POLL_MS = 25_000
export const ROLE_FAST_WINDOW_MS = 3 * 60_000 export const ROLE_SLOW_POLL_MS = 45_000
export const ROLE_IDLE_POLL_MS = 90_000
/** 快轮询窗口:原 3 分钟 → 2 分钟 */
export const ROLE_FAST_WINDOW_MS = 2 * 60_000
export const ROLE_IDLE_WINDOW_MS = 10 * 60_000 export const ROLE_IDLE_WINDOW_MS = 10 * 60_000
export const FEIFEI_POLL_MS = 15_000 export const FEIFEI_POLL_MS = 20_000
export function resolveRolePollDelayMs( export function resolveRolePollDelayMs(
nextSnapshot: ClaimSnapshot, nextSnapshot: ClaimSnapshot,
@@ -35,6 +38,11 @@ export function resolveRolePollDelayMs(
delay = ROLE_SLOW_POLL_MS delay = ROLE_SLOW_POLL_MS
} }
// 已拿到角色但未匹配:不必高频刷,稍放缓即可
if (nextSnapshot.roleId && !nextSnapshot.isUidMatched) {
delay = Math.max(delay, ROLE_SLOW_POLL_MS)
}
return { return {
delay, delay,
baseline: { roleKey: baselineKey, at: baselineAt }, baseline: { roleKey: baselineKey, at: baselineAt },