From 681ff44cbcc3d0632db608eca1c83c56e9703d6a Mon Sep 17 00:00:00 2001 From: yml2213 Date: Fri, 29 May 2026 23:21:32 +0800 Subject: [PATCH] =?UTF-8?q?=E4=BC=98=E5=8C=96high-frequency=20Request=20?= =?UTF-8?q?=E9=97=AE=E9=A2=98?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../cloudtentacles/http-client.test.ts | 127 ++++++++ .../platforms/cloudtentacles/http-client.ts | 272 ++++++++++++++---- 2 files changed, 343 insertions(+), 56 deletions(-) create mode 100644 apps/backend/src/services/platforms/cloudtentacles/http-client.test.ts diff --git a/apps/backend/src/services/platforms/cloudtentacles/http-client.test.ts b/apps/backend/src/services/platforms/cloudtentacles/http-client.test.ts new file mode 100644 index 00000000..e55951c4 --- /dev/null +++ b/apps/backend/src/services/platforms/cloudtentacles/http-client.test.ts @@ -0,0 +1,127 @@ +import test from 'node:test' +import assert from 'node:assert/strict' +import http from 'node:http' + +import { cloudtentaclesRequest } from './http-client.js' + +type MockResponse = { + status?: number + body?: unknown +} + +test('cloudtentaclesRequest retries high-frequency business errors', async () => { + let callCount = 0 + const server = await createMockServer(() => { + callCount += 1 + return { + body: callCount === 1 + ? { code: 1, message: 'high-frequency Request' } + : { code: 0, data: 'ok' }, + } + }) + + try { + const result = await cloudtentaclesRequest('/user/get_asset', { + baseUrl: server.baseUrl, + token: 'retry-token', + sourceKey: '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 () => { + const server = await createMockServer(() => ({ + body: { code: 1, message: 'high-frequency Request' }, + })) + + try { + await assert.rejects( + cloudtentaclesRequest('/user/get_asset', { + baseUrl: server.baseUrl, + token: 'limited-token', + sourceKey: 'limited-source', + rateLimitIntervalMs: 0, + rateLimitRetries: 0, + }), + (error) => { + assert.equal(error instanceof Error, true) + assert.equal(Reflect.get(error as object, 'statusCode'), 429) + assert.equal(Reflect.get(error as object, 'errorCode'), 'cloudtentacles_rate_limited') + return true + }, + ) + } finally { + await server.close() + } +}) + +test('cloudtentaclesRequest spaces concurrent requests in the same bucket', async () => { + const requestTimes: number[] = [] + const server = await createMockServer(() => { + requestTimes.push(Date.now()) + return { body: { code: 0, data: 'ok' } } + }) + + try { + await Promise.all([ + cloudtentaclesRequest('/sku/list', { + baseUrl: server.baseUrl, + token: 'throttle-token', + sourceKey: 'throttle-source', + rateLimitIntervalMs: 25, + rateLimitRetries: 0, + }), + cloudtentaclesRequest('/user/get_knapsack', { + baseUrl: server.baseUrl, + token: 'throttle-token', + sourceKey: 'throttle-source', + rateLimitIntervalMs: 25, + rateLimitRetries: 0, + }), + ]) + + assert.equal(requestTimes.length, 2) + assert.ok(Math.abs(Number(requestTimes[1]) - Number(requestTimes[0])) >= 20) + } finally { + await server.close() + } +}) + +async function createMockServer(handler: () => MockResponse) { + const server = http.createServer((_request, response) => { + const result = handler() + response.statusCode = result.status || 200 + response.setHeader('content-type', 'application/json') + response.end(JSON.stringify(result.body ?? { code: 0, data: null })) + }) + + await new Promise((resolve) => { + server.listen(0, '127.0.0.1', resolve) + }) + + const address = server.address() + assert.equal(typeof address, 'object') + assert.notEqual(address, null) + + return { + baseUrl: `http://127.0.0.1:${address?.port}`, + close: () => new Promise((resolve, reject) => { + server.close((error) => { + if (error) { + reject(error) + return + } + + resolve() + }) + }), + } +} diff --git a/apps/backend/src/services/platforms/cloudtentacles/http-client.ts b/apps/backend/src/services/platforms/cloudtentacles/http-client.ts index c90fad65..bec85330 100644 --- a/apps/backend/src/services/platforms/cloudtentacles/http-client.ts +++ b/apps/backend/src/services/platforms/cloudtentacles/http-client.ts @@ -2,7 +2,7 @@ import http from 'node:http' import https from 'node:https' import { createHttpError } from '../../../utils/http.js' -import { logInfo } from '../../../utils/logger.js' +import { logInfo, logWarn } from '../../../utils/logger.js' import { notifyCloudtentaclesAuthExpired } from '../../notification/domain-notifications.js' import { buildCloudtentaclesHeaders, buildCloudtentaclesUrl, resolveCloudtentaclesConfig } from './helpers.js' @@ -17,13 +17,18 @@ type NodeHttpResponse = { bodyText: string } +const DEFAULT_RATE_LIMIT_INTERVAL_MS = 300 +const DEFAULT_RATE_LIMIT_RETRIES = 2 +const DEFAULT_RATE_LIMIT_RETRY_DELAY_MS = 800 +const MAX_RETRY_JITTER_MS = 180 +const bucketNextAvailableAt = new Map() +let rateLimitQueue = Promise.resolve() + export async function cloudtentaclesRequest(pathname: unknown, options: JsonObject = {}) { const normalizedPathname = String(pathname || '').trim() const config = resolveCloudtentaclesConfig(options) const url = buildCloudtentaclesUrl(config.baseUrl, normalizedPathname, options.searchParams) const timeoutMs = Number(options.timeoutMs || config.timeoutMs || 5000) - const controller = new AbortController() - const timer = setTimeout(() => controller.abort(), timeoutMs) const method = String(options.method || 'GET').trim().toUpperCase() const headers = buildCloudtentaclesHeaders({ token: options.token, @@ -32,71 +37,111 @@ export async function cloudtentaclesRequest(pathname: unknown, options: JsonObje deviceType: options.deviceType ?? config.deviceType, extra: options.headers, }) + const rateLimitBucketKey = resolveRateLimitBucketKey({ + baseUrl: config.baseUrl, + token: options.token, + sourceKey: options.sourceKey, + deviceId: options.deviceId ?? config.deviceId, + }) + const rateLimitIntervalMs = normalizeNonNegativeInteger(options.rateLimitIntervalMs, DEFAULT_RATE_LIMIT_INTERVAL_MS) + const maxRetries = normalizeNonNegativeInteger(options.rateLimitRetries, DEFAULT_RATE_LIMIT_RETRIES) + const retryDelayMs = normalizeNonNegativeInteger(options.rateLimitRetryDelayMs, DEFAULT_RATE_LIMIT_RETRY_DELAY_MS) - try { - const body = normalizeRequestBody(options.body, headers['content-type']) - const response = await requestViaNodeHttp(url, { - method, - headers, - signal: controller.signal, - ...(body !== undefined ? { body } : {}), - }) - - const rawText = response.bodyText - const payload = tryParseJson(rawText) - - if (!response.ok) { - throw createHttpError(`cloudtentacles 请求失败，HTTP ${response.status}`, { - statusCode: 502, - errorCode: 'cloudtentacles_http_error', + for (let attempt = 0; attempt <= maxRetries; attempt += 1) { + try { + await waitForCloudtentaclesSlot(rateLimitBucketKey, rateLimitIntervalMs) + const body = normalizeRequestBody(options.body, headers['content-type']) + const response = await executeCloudtentaclesRequest({ + url, + method, + headers, + timeoutMs, + ...(body !== undefined ? { body } : {}), }) - } + const rawText = response.bodyText + const payload = tryParseJson(rawText) - if (options.requireBusinessSuccess !== false && Number(payload?.code ?? 1) !== 0) { - const message = String(payload?.message || payload?.msg || 'cloudtentacles 接口返回失败') - const statusCode = Number(options.businessErrorStatusCode || 400) - const errorCode = String(options.businessErrorCode || 'cloudtentacles_business_error') + if (!response.ok) { + if (isCloudtentaclesHttpRateLimited(response.status) && attempt < maxRetries) { + await waitBeforeRateLimitRetry(attempt, normalizedPathname, method, response.status, retryDelayMs) + continue + } - if (isCloudtentaclesExpiredMessage(message)) { - await notifyCloudtentaclesAuthExpired({ - pathname: normalizedPathname, - errorCode, - message, - sourceKey: options.sourceKey, - accountLabel: options.accountLabel, + throw createHttpError(`cloudtentacles 请求失败，HTTP ${response.status}`, { + statusCode: isCloudtentaclesHttpRateLimited(response.status) ? 429 : 502, + errorCode: isCloudtentaclesHttpRateLimited(response.status) + ? 'cloudtentacles_rate_limited' + : 'cloudtentacles_http_error', }) } - throw createHttpError(message, { - statusCode, - errorCode, - }) - } + if (options.requireBusinessSuccess !== false && Number(payload?.code ?? 1) !== 0) { + const message = String(payload?.message || payload?.msg || 'cloudtentacles 接口返回失败') + const statusCode = Number(options.businessErrorStatusCode || 400) + const errorCode = String(options.businessErrorCode || 'cloudtentacles_business_error') - logInfo('[cloudtentacles/http]', '请求完成', { - method, - pathname: normalizedPathname, - status: response.status, - }) + if (isHighFrequencyMessage(message)) { + if (attempt < maxRetries) { + await waitBeforeRateLimitRetry(attempt, normalizedPathname, method, response.status, retryDelayMs) + continue + } - return { - url: url.toString(), - status: response.status, - headers: response.headers, - payload, - rawText, - } - } catch (error) { - if (error instanceof Error && error.name === 'AbortError') { - throw createHttpError(`cloudtentacles 请求超时（${timeoutMs}ms）`, { - statusCode: 504, - errorCode: 'cloudtentacles_request_timeout', + throw createHttpError('cloudtentacles 请求过于频繁，请稍后重试', { + statusCode: 429, + errorCode: 'cloudtentacles_rate_limited', + context: { + pathname: normalizedPathname, + upstreamMessage: message, + }, + }) + } + + if (isCloudtentaclesExpiredMessage(message)) { + await notifyCloudtentaclesAuthExpired({ + pathname: normalizedPathname, + errorCode, + message, + sourceKey: options.sourceKey, + accountLabel: options.accountLabel, + }) + } + + throw createHttpError(message, { + statusCode, + errorCode, + }) + } + + logInfo('[cloudtentacles/http]', '请求完成', { + method, + pathname: normalizedPathname, + status: response.status, + attempt: attempt + 1, }) + + return { + url: url.toString(), + status: response.status, + headers: response.headers, + payload, + rawText, + } + } catch (error) { + if (error instanceof Error && error.name === 'AbortError') { + throw createHttpError(`cloudtentacles 请求超时（${timeoutMs}ms）`, { + statusCode: 504, + errorCode: 'cloudtentacles_request_timeout', + }) + } + + throw error } - throw error - } finally { - clearTimeout(timer) } + + throw createHttpError('cloudtentacles 请求过于频繁，请稍后重试', { + statusCode: 429, + errorCode: 'cloudtentacles_rate_limited', + }) } function isCloudtentaclesExpiredMessage(message: unknown) { @@ -104,6 +149,121 @@ function isCloudtentaclesExpiredMessage(message: unknown) { 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) { + return Number(status) === 429 +} + +async function executeCloudtentaclesRequest({ + url, + method, + headers, + body, + timeoutMs, +}: { + url: URL + method: string + headers: JsonObject + body?: string | URLSearchParams + timeoutMs: number +}) { + const controller = new AbortController() + const timer = setTimeout(() => controller.abort(), timeoutMs) + + try { + return await requestViaNodeHttp(url, { + method, + headers, + signal: controller.signal, + ...(body !== undefined ? { body } : {}), + }) + } finally { + clearTimeout(timer) + } +} + +async function waitForCloudtentaclesSlot(bucketKey: string, intervalMs: number) { + if (intervalMs <= 0) { + return + } + + const previousQueue = rateLimitQueue + let releaseQueue: () => void = () => {} + rateLimitQueue = new Promise((resolve) => { + releaseQueue = resolve + }) + + await previousQueue + + let waitMs = 0 + try { + const now = Date.now() + const nextAvailableAt = bucketNextAvailableAt.get(bucketKey) || 0 + waitMs = Math.max(0, nextAvailableAt - now) + bucketNextAvailableAt.set(bucketKey, Math.max(now, nextAvailableAt) + intervalMs) + } finally { + releaseQueue() + } + + if (waitMs > 0) { + await sleep(waitMs) + } +} + +async function waitBeforeRateLimitRetry( + attempt: number, + pathname: string, + method: string, + status: number, + retryDelayMs: number, +) { + const delayMs = retryDelayMs * (attempt + 1) + Math.floor(Math.random() * MAX_RETRY_JITTER_MS) + logWarn('[cloudtentacles/http]', '请求触发云触手限频，准备重试', { + method, + pathname, + status, + attempt: attempt + 1, + retryDelayMs: delayMs, + }) + await sleep(delayMs) +} + +function resolveRateLimitBucketKey({ + baseUrl, + token, + sourceKey, + deviceId, +}: { + baseUrl: unknown + token: unknown + sourceKey: unknown + deviceId: unknown +}) { + return [ + String(baseUrl || '').trim(), + String(sourceKey || '').trim() || String(token || '').trim() || 'anonymous', + String(deviceId || '').trim(), + ].join('|') +} + +function normalizeNonNegativeInteger(value: unknown, fallback: number) { + const numericValue = Number(value) + if (Number.isInteger(numericValue) && numericValue >= 0) { + return numericValue + } + + return fallback +} + +function sleep(ms: number) { + return new Promise((resolve) => { + setTimeout(resolve, ms) + }) +} + function inferContentType(body: unknown) { if (body == null) { return ''