彻底修复 lewan 自动发货问题-8-8

2026-08-08 23:08:05 +08:00
parent 7196752d44
commit 6e41d7e0db
10 changed files with 260 additions and 37 deletions
@@ -20,12 +20,21 @@ const contexts = {
  },
 }

+const listEmptyVirtualNumbers = async () => ({
+  baseUrl: 'https://cloud.example.com',
+  key: '1',
+  itemCount: 0,
+  items: [],
+  rawItems: [],
+})
+
 test('selectCloudtentaclesSourceForFulfillment 已固定账号时不按压力切换', async () => {
  const selection = await selectCloudtentaclesSourceForFulfillment(
    {
      binding: {
        resolvedSourceKey: 'account-a',
        cloudSourceKeys: ['account-a', 'account-b'],
+        vnId: 123,
      },
    },
    {
@@ -37,6 +46,7 @@ test('selectCloudtentaclesSourceForFulfillment 已固定账号时不按压力切
        { sourceKey: 'account-a', activeCount: 99, redeemingCount: 99 },
        { sourceKey: 'account-b', activeCount: 0, redeemingCount: 0 },
      ],
+      listVirtualNumbers: listEmptyVirtualNumbers,
    },
  )

@@ -44,6 +54,61 @@ test('selectCloudtentaclesSourceForFulfillment 已固定账号时不按压力切
  assert.equal(selection.fixed, true)
 })

+test('selectCloudtentaclesSourceForFulfillment 无号码时忽略残留固定账号', async () => {
+  const selection = await selectCloudtentaclesSourceForFulfillment(
+    {
+      binding: {
+        resolvedSourceKey: 'account-a',
+        cloudSourceKeys: ['account-a', 'account-b'],
+        vnId: 0,
+      },
+    },
+    {
+      resolveContextBySourceKeys: (sourceKeys) => contexts[String(sourceKeys[0]) as keyof typeof contexts],
+      listSourceLoadStats: async () => [
+        { sourceKey: 'account-a', activeCount: 20, redeemingCount: 0 },
+        { sourceKey: 'account-b', activeCount: 1, redeemingCount: 0 },
+      ],
+      listVirtualNumbers: listEmptyVirtualNumbers,
+    },
+  )
+
+  try {
+    assert.equal(selection.sourceKey, 'account-b')
+    assert.equal(selection.fixed, false)
+  } finally {
+    selection.release()
+  }
+})
+
+test('selectCloudtentaclesSourceForFulfillment 跳过真实占用已达20的账号', async () => {
+  const selection = await selectCloudtentaclesSourceForFulfillment(
+    { binding: { cloudSourceKeys: ['account-a', 'account-b'] } },
+    {
+      resolveContextBySourceKeys: (sourceKeys) => contexts[String(sourceKeys[0]) as keyof typeof contexts],
+      listSourceLoadStats: async () => [],
+      listVirtualNumbers: async (payload) => {
+        const count = payload.sourceKey === 'account-a' ? 20 : 3
+        return {
+          baseUrl: 'https://cloud.example.com',
+          key: '1',
+          itemCount: count,
+          items: Array.from({ length: count }, (_, id) => ({ id: id + 1 })),
+          rawItems: [],
+        }
+      },
+    },
+  )
+
+  try {
+    assert.equal(selection.sourceKey, 'account-b')
+    assert.equal(selection.occupiedCount, 3)
+    assert.equal(selection.availableCount, 17)
+  } finally {
+    selection.release()
+  }
+})
+
 test('selectCloudtentaclesSourceForFulfillment 多候选时选择压力最低账号', async () => {
  const selection = await selectCloudtentaclesSourceForFulfillment(
    {
@@ -60,6 +125,7 @@ test('selectCloudtentaclesSourceForFulfillment 多候选时选择压力最低账
        { sourceKey: 'account-a', activeCount: 3, redeemingCount: 1 },
        { sourceKey: 'account-b', activeCount: 0, redeemingCount: 0 },
      ],
+      listVirtualNumbers: listEmptyVirtualNumbers,
    },
  )

@@ -87,6 +153,7 @@ test('selectCloudtentaclesSourceForFulfillment 会跳过不可用账号', async
        return contexts[key as keyof typeof contexts]
      },
      listSourceLoadStats: async () => [],
+      listVirtualNumbers: listEmptyVirtualNumbers,
    },
  )

@@ -105,6 +172,7 @@ test('selectCloudtentaclesSourceForFulfillment 用进程内占位分散并发准
      return contexts[key as keyof typeof contexts]
    },
    listSourceLoadStats: async () => [],
+    listVirtualNumbers: listEmptyVirtualNumbers,
  }
  const flow = {
    binding: {
@@ -1,12 +1,20 @@
 import { listKuaishouCloudSourceLoadStats } from '../../../repositories/kuaishou-cloud-task-state-repo.js'
-import { normalizeKuaishouCloudFlow, normalizeStringArray, type JsonObject } from './domain.js'
+import {
+  KUAISHOU_CLOUD_FIXED_VN_KEY,
+  normalizeKuaishouCloudFlow,
+  normalizeStringArray,
+  type JsonObject,
+} from './domain.js'
 import { resolvePersistedCloudtentaclesContextWithFallback } from './cloudtentacles-context.js'
+import { listCloudtentaclesVirtualNumbers } from '../../platforms/cloudtentacles/virtual-number-service.js'
+import { createHttpError } from '../../../utils/http.js'

 type CloudtentaclesContext = ReturnType<typeof resolvePersistedCloudtentaclesContextWithFallback>

 type CloudtentaclesSourceSelectionDeps = {
  resolveContextBySourceKeys?: typeof resolvePersistedCloudtentaclesContextWithFallback
  listSourceLoadStats?: typeof listKuaishouCloudSourceLoadStats
+  listVirtualNumbers?: typeof listCloudtentaclesVirtualNumbers
 }

 export type CloudtentaclesSourceSelection = {
@@ -17,6 +25,9 @@ export type CloudtentaclesSourceSelection = {
  redeemingCount: number
  inProcessCount: number
  score: number
+  occupiedCount: number
+  capacity: number
+  availableCount: number
  fixed: boolean
  release: () => void
 }
@@ -32,12 +43,15 @@ export async function selectCloudtentaclesSourceForFulfillment(
  const resolveContext =
    deps.resolveContextBySourceKeys || resolvePersistedCloudtentaclesContextWithFallback
  const listSourceLoadStats = deps.listSourceLoadStats || listKuaishouCloudSourceLoadStats
-  const fixedSourceKey = String(flow.binding.resolvedSourceKey || '').trim()
+  const listVirtualNumbers = deps.listVirtualNumbers || listCloudtentaclesVirtualNumbers
+  const persistedSourceKey = String(flow.binding.resolvedSourceKey || '').trim()
+  const fixedSourceKey = flow.binding.vnId > 0 ? persistedSourceKey : ''
+  const excludedSourceKeys = new Set(normalizeStringArray((flow as JsonObject).excludedSourceKeys))
  const candidates = normalizeCloudtentaclesSourceCandidates(
    fixedSourceKey
      ? [fixedSourceKey, ...flow.binding.cloudSourceKeys]
-      : flow.binding.cloudSourceKeys,
-  )
+      : [...flow.binding.cloudSourceKeys, persistedSourceKey],
+  ).filter((key) => !excludedSourceKeys.has(key))

  if (fixedSourceKey) {
    const context = resolveContext([fixedSourceKey, ...flow.binding.cloudSourceKeys])
@@ -49,6 +63,9 @@ export async function selectCloudtentaclesSourceForFulfillment(
      redeemingCount: 0,
      inProcessCount: 0,
      score: 0,
+      occupiedCount: 0,
+      capacity: 20,
+      availableCount: 20,
      fixed: true,
      release: () => {},
    }
@@ -76,40 +93,66 @@ export async function selectCloudtentaclesSourceForFulfillment(

    const stats = await listSourceLoadStats(contexts.map((context) => context.resolvedSourceKey))
    const statsBySourceKey = new Map(stats.map((stat) => [stat.sourceKey, stat]))
-    const ranked = contexts
-      .map((context, index) => {
-        const stat = statsBySourceKey.get(context.resolvedSourceKey) || {
+    const ranked = []
+    let lastCapacityError: unknown = null
+    for (const [index, context] of contexts.entries()) {
+      let occupiedCount = 0
+      try {
+        // CT's VN list is the authoritative count; DB counters can lag after crashes/restarts.
+        const listed = await listVirtualNumbers({
+          ...context,
+          key: KUAISHOU_CLOUD_FIXED_VN_KEY,
          sourceKey: context.resolvedSourceKey,
-          activeCount: 0,
-          redeemingCount: 0,
-        }
-        const activeCount = normalizeCount(stat.activeCount)
-        const redeemingCount = normalizeCount(stat.redeemingCount)
-        const inProcessCount = normalizeCount(inProcessSelections.get(context.resolvedSourceKey))
-        const score = (activeCount + inProcessCount) * 10 + redeemingCount * 20
+          accountLabel: context.accountLabel,
+        })
+        occupiedCount = Array.isArray(listed.items) ? listed.items.length : Number(listed.itemCount || 0) || 0
+      } catch (error) {
+        lastCapacityError = error
+        continue
+      }
+      const capacity = 20
+      const inProcessCount = normalizeCount(inProcessSelections.get(context.resolvedSourceKey))
+      if (occupiedCount + inProcessCount >= capacity) continue
+      const stat = statsBySourceKey.get(context.resolvedSourceKey) || {
+        sourceKey: context.resolvedSourceKey,
+        activeCount: 0,
+        redeemingCount: 0,
+      }
+      const activeCount = normalizeCount(stat.activeCount)
+      const redeemingCount = normalizeCount(stat.redeemingCount)
+      // Real upstream occupancy is the primary balancing signal. DB state is
+      // retained as a secondary tie-breaker for tasks still being processed.
+      const score = (occupiedCount + inProcessCount) * 100 + activeCount * 10 + redeemingCount * 20

-        return {
-          context,
-          sourceKey: context.resolvedSourceKey,
-          index,
-          activeCount,
-          redeemingCount,
-          inProcessCount,
-          score,
-        }
+      ranked.push({
+        context,
+        sourceKey: context.resolvedSourceKey,
+        index,
+        activeCount,
+        redeemingCount,
+        inProcessCount,
+        score,
+        occupiedCount,
+        capacity,
+        availableCount: Math.max(0, capacity - occupiedCount - inProcessCount),
      })
-      .sort(
-        (left, right) =>
-          left.score - right.score ||
-          left.activeCount - right.activeCount ||
-          left.redeemingCount - right.redeemingCount ||
-          left.index - right.index ||
-          left.sourceKey.localeCompare(right.sourceKey),
-      )
+    }
+    if (ranked.length === 0 && lastCapacityError && contexts.length > 0) throw lastCapacityError
+    ranked.sort(
+      (left, right) =>
+        left.score - right.score ||
+        left.activeCount - right.activeCount ||
+        left.redeemingCount - right.redeemingCount ||
+        left.index - right.index ||
+        left.sourceKey.localeCompare(right.sourceKey),
+    )

    const selected = ranked[0]
    if (!selected) {
-      throw new Error('所有 cloudtentacles 账号均不可用')
+      throw createHttpError('所有 cloudtentacles 账号虚拟号配额均已满', {
+        statusCode: 409,
+        errorCode: 'cloudtentacles_all_sources_at_capacity',
+      })
    }

    reserveCloudtentaclesSource(selected.sourceKey)
@@ -122,6 +165,9 @@ export async function selectCloudtentaclesSourceForFulfillment(
      redeemingCount: selected.redeemingCount,
      inProcessCount: selected.inProcessCount,
      score: selected.score,
+      occupiedCount: selected.occupiedCount,
+      capacity: selected.capacity,
+      availableCount: selected.availableCount,
      fixed: false,
      release: () => releaseCloudtentaclesSource(selected.sourceKey),
    }
@@ -138,6 +184,9 @@ export function buildCloudtentaclesSourceSelectionEventDetail(
    redeemingCount: selection.redeemingCount,
    inProcessCount: selection.inProcessCount,
    score: selection.score,
+    occupiedCount: selection.occupiedCount,
+    capacity: selection.capacity,
+    availableCount: selection.availableCount,
    fixed: selection.fixed,
  }
 }
@@ -18,6 +18,7 @@ import {
  isKuaishouCloudTask,
  maskPhone,
  normalizeKuaishouCloudFlow,
+  normalizeStringArray,
  resolveKuaishouCloudBindUrlExpiresAt,
  type JsonObject,
 } from './domain.js'
@@ -123,7 +124,16 @@ export async function prepareKuaishouCloudFulfillmentTask(task: TaskRow, options
    })
  }

-  const selectedCloud = await selectCloudtentaclesSourceForFulfillment(flow)
+  const excludedSourceKeys = normalizeStringArray(options.excludedSourceKeys)
+  const sourceSelectionAttempts = normalizeStringArray(options.sourceSelectionAttempts)
+  const sourceCandidateCount = new Set(
+    normalizeStringArray([...flow.binding.cloudSourceKeys, flow.binding.resolvedSourceKey]),
+  ).size
+  const selectedCloud = await selectCloudtentaclesSourceForFulfillment({
+    ...flow,
+    excludedSourceKeys,
+  })
+  let retriedOnAnotherSource = false
  try {
    const cloudContext = selectedCloud.context
    const [knapsack, skuList] = await Promise.all([
@@ -277,7 +287,10 @@ export async function prepareKuaishouCloudFulfillmentTask(task: TaskRow, options
          purchasedCount: 0,
        })),
        resolvedByName: resolvedBinding.resolvedByName,
-        accountSelection: buildCloudtentaclesSourceSelectionEventDetail(selectedCloud),
+        accountSelection: {
+          ...buildCloudtentaclesSourceSelectionEventDetail(selectedCloud),
+          attemptedSourceKeys: [...sourceSelectionAttempts, selectedCloud.sourceKey],
+        },
        defaultRoleName: defaultRoleSnapshot.defaultName,
        defaultRoleId: defaultRoleSnapshot.defaultRid,
        defaultRoleCaptureStatus: defaultRoleSnapshot.defaultCaptureStatus,
@@ -293,11 +306,43 @@ export async function prepareKuaishouCloudFulfillmentTask(task: TaskRow, options
      flow: normalizeKuaishouCloudFlow(nextContext.kuaishouCloudFulfillment),
      previousVnRelease,
    }
+  } catch (error) {
+    // Before a VN exists, account-scoped failures are safe to retry on another
+    // candidate. Once a VN is acquired, switching accounts would orphan it.
+    if (
+      flow.binding.vnId <= 0 &&
+      isRetryableCloudtentaclesSourceError(error) &&
+      excludedSourceKeys.length + 1 < sourceCandidateCount
+    ) {
+      retriedOnAnotherSource = true
+      selectedCloud.release()
+      return prepareKuaishouCloudFulfillmentTask(task, {
+        ...options,
+        excludedSourceKeys: [...excludedSourceKeys, selectedCloud.sourceKey],
+        sourceSelectionAttempts: [...sourceSelectionAttempts, selectedCloud.sourceKey],
+      })
+    }
+    throw error
  } finally {
-    selectedCloud.release()
+    if (!retriedOnAnotherSource) {
+      selectedCloud.release()
+    }
  }
 }

+function isRetryableCloudtentaclesSourceError(error: unknown) {
+  const current = error && typeof error === 'object' ? error as JsonObject : {}
+  const code = String(current.errorCode || current.code || '').trim()
+  return (
+    code === 'cloudtentacles_vn_quota_cooldown' ||
+    (code === 'cloudtentacles_vn_appoint_failed' && String(current.message || '').trim().includes('最多同时占用')) ||
+    code === 'cloudtentacles_vn_list_failed' ||
+    code === 'cloudtentacles_sku_list_failed' ||
+    code === 'cloudtentacles_knapsack_failed' ||
+    code === 'cloudtentacles_asset_failed'
+  )
+}
+
 /** 旧虚拟号 best-effort 释放：已被回收/权限不足时不算失败 */
 async function tryReleasePreviousVirtualNumber({
  task,