优化high-frequency Request 问题

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<void>((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<void>((resolve, reject) => {
+      server.close((error) => {
+        if (error) {
+          reject(error)
+          return
+        }
+
+        resolve()
+      })
+    }),
+  }
+}

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<string, number>()
+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 {
+  for (let attempt = 0; attempt <= maxRetries; attempt += 1) {
-    const body = normalizeRequestBody(options.body, headers['content-type'])
+    try {
-    const response = await requestViaNodeHttp(url, {
+      await waitForCloudtentaclesSlot(rateLimitBucketKey, rateLimitIntervalMs)
-      method,
+      const body = normalizeRequestBody(options.body, headers['content-type'])
-      headers,
+      const response = await executeCloudtentaclesRequest({
-      signal: controller.signal,
+        url,
-      ...(body !== undefined ? { body } : {}),
+        method,
-    })
+        headers,
-
+        timeoutMs,
-    const rawText = response.bodyText
+        ...(body !== undefined ? { body } : {}),
-    const payload = tryParseJson(rawText)
-
-    if (!response.ok) {
-      throw createHttpError(`cloudtentacles 请求失败，HTTP ${response.status}`, {
-        statusCode: 502,
-        errorCode: 'cloudtentacles_http_error',
       })
-    }
+      const rawText = response.bodyText
+      const payload = tryParseJson(rawText)
 
-    if (options.requireBusinessSuccess !== false && Number(payload?.code ?? 1) !== 0) {
+      if (!response.ok) {
-      const message = String(payload?.message || payload?.msg || 'cloudtentacles 接口返回失败')
+        if (isCloudtentaclesHttpRateLimited(response.status) && attempt < maxRetries) {
-      const statusCode = Number(options.businessErrorStatusCode || 400)
+          await waitBeforeRateLimitRetry(attempt, normalizedPathname, method, response.status, retryDelayMs)
-      const errorCode = String(options.businessErrorCode || 'cloudtentacles_business_error')
+          continue
+        }
 
-      if (isCloudtentaclesExpiredMessage(message)) {
+        throw createHttpError(`cloudtentacles 请求失败，HTTP ${response.status}`, {
-        await notifyCloudtentaclesAuthExpired({
+          statusCode: isCloudtentaclesHttpRateLimited(response.status) ? 429 : 502,
-          pathname: normalizedPathname,
+          errorCode: isCloudtentaclesHttpRateLimited(response.status)
-          errorCode,
+            ? 'cloudtentacles_rate_limited'
-          message,
+            : 'cloudtentacles_http_error',
-          sourceKey: options.sourceKey,
-          accountLabel: options.accountLabel,
         })
       }
 
-      throw createHttpError(message, {
+      if (options.requireBusinessSuccess !== false && Number(payload?.code ?? 1) !== 0) {
-        statusCode,
+        const message = String(payload?.message || payload?.msg || 'cloudtentacles 接口返回失败')
-        errorCode,
+        const statusCode = Number(options.businessErrorStatusCode || 400)
-      })
+        const errorCode = String(options.businessErrorCode || 'cloudtentacles_business_error')
-    }
 
-    logInfo('[cloudtentacles/http]', '请求完成', {
+        if (isHighFrequencyMessage(message)) {
-      method,
+          if (attempt < maxRetries) {
-      pathname: normalizedPathname,
+            await waitBeforeRateLimitRetry(attempt, normalizedPathname, method, response.status, retryDelayMs)
-      status: response.status,
+            continue
-    })
+          }
 
-    return {
+          throw createHttpError('cloudtentacles 请求过于频繁，请稍后重试', {
-      url: url.toString(),
+            statusCode: 429,
-      status: response.status,
+            errorCode: 'cloudtentacles_rate_limited',
-      headers: response.headers,
+            context: {
-      payload,
+              pathname: normalizedPathname,
-      rawText,
+              upstreamMessage: message,
-    }
+            },
-  } catch (error) {
+          })
-    if (error instanceof Error && error.name === 'AbortError') {
+        }
-      throw createHttpError(`cloudtentacles 请求超时（${timeoutMs}ms）`, {
+
-        statusCode: 504,
+        if (isCloudtentaclesExpiredMessage(message)) {
-        errorCode: 'cloudtentacles_request_timeout',
+          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<void>((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 ''