优化多账号选择

2026-05-29 22:34:41 +08:00
parent e5ad6f4918
commit f408fe1804
6 changed files with 1006 additions and 517 deletions
@@ -0,0 +1,124 @@
+import assert from 'node:assert/strict'
+import test from 'node:test'
+
+import { selectCloudtentaclesSourceForFulfillment } from './account-selector.js'
+
+const contexts = {
+  'account-a': {
+    baseUrl: 'https://cloud.example.com',
+    token: 'token-a',
+    deviceId: '-',
+    deviceType: 0,
+    resolvedSourceKey: 'account-a',
+  },
+  'account-b': {
+    baseUrl: 'https://cloud.example.com',
+    token: 'token-b',
+    deviceId: '-',
+    deviceType: 0,
+    resolvedSourceKey: 'account-b',
+  },
+}
+
+test('selectCloudtentaclesSourceForFulfillment 已固定账号时不按压力切换', async () => {
+  const selection = await selectCloudtentaclesSourceForFulfillment(
+    {
+      binding: {
+        resolvedSourceKey: 'account-a',
+        cloudSourceKeys: ['account-a', 'account-b'],
+      },
+    },
+    {
+      resolveContextBySourceKeys: (sourceKeys) => {
+        const key = String(sourceKeys[0] || '').trim()
+        return contexts[key as keyof typeof contexts]
+      },
+      listSourceLoadStats: async () => [
+        { sourceKey: 'account-a', activeCount: 99, redeemingCount: 99 },
+        { sourceKey: 'account-b', activeCount: 0, redeemingCount: 0 },
+      ],
+    },
+  )
+
+  assert.equal(selection.sourceKey, 'account-a')
+  assert.equal(selection.fixed, true)
+})
+
+test('selectCloudtentaclesSourceForFulfillment 多候选时选择压力最低账号', async () => {
+  const selection = await selectCloudtentaclesSourceForFulfillment(
+    {
+      binding: {
+        cloudSourceKeys: ['account-a', 'account-b'],
+      },
+    },
+    {
+      resolveContextBySourceKeys: (sourceKeys) => {
+        const key = String(sourceKeys[0] || '').trim()
+        return contexts[key as keyof typeof contexts]
+      },
+      listSourceLoadStats: async () => [
+        { sourceKey: 'account-a', activeCount: 3, redeemingCount: 1 },
+        { sourceKey: 'account-b', activeCount: 0, redeemingCount: 0 },
+      ],
+    },
+  )
+
+  try {
+    assert.equal(selection.sourceKey, 'account-b')
+    assert.equal(selection.fixed, false)
+  } finally {
+    selection.release()
+  }
+})
+
+test('selectCloudtentaclesSourceForFulfillment 会跳过不可用账号', async () => {
+  const selection = await selectCloudtentaclesSourceForFulfillment(
+    {
+      binding: {
+        cloudSourceKeys: ['account-a', 'account-b'],
+      },
+    },
+    {
+      resolveContextBySourceKeys: (sourceKeys) => {
+        const key = String(sourceKeys[0] || '').trim()
+        if (key === 'account-a') {
+          throw new Error('账号已停用')
+        }
+        return contexts[key as keyof typeof contexts]
+      },
+      listSourceLoadStats: async () => [],
+    },
+  )
+
+  try {
+    assert.equal(selection.sourceKey, 'account-b')
+    assert.equal(selection.candidateCount, 1)
+  } finally {
+    selection.release()
+  }
+})
+
+test('selectCloudtentaclesSourceForFulfillment 用进程内占位分散并发准备', async () => {
+  const deps = {
+    resolveContextBySourceKeys: (sourceKeys: unknown[] = []) => {
+      const key = String(sourceKeys[0] || '').trim()
+      return contexts[key as keyof typeof contexts]
+    },
+    listSourceLoadStats: async () => [],
+  }
+  const flow = {
+    binding: {
+      cloudSourceKeys: ['account-a', 'account-b'],
+    },
+  }
+  const first = await selectCloudtentaclesSourceForFulfillment(flow, deps)
+  const second = await selectCloudtentaclesSourceForFulfillment(flow, deps)
+
+  try {
+    assert.equal(first.sourceKey, 'account-a')
+    assert.equal(second.sourceKey, 'account-b')
+  } finally {
+    first.release()
+    second.release()
+  }
+})
@@ -0,0 +1,182 @@
+import { listKuaishouCloudSourceLoadStats } from '../../../repositories/kuaishou-cloud-task-state-repo.js'
+import { normalizeKuaishouCloudFlow, normalizeStringArray, type JsonObject } from './domain.js'
+import { resolvePersistedCloudtentaclesContextBySourceKeys } from './cloudtentacles-context.js'
+
+type CloudtentaclesContext = ReturnType<typeof resolvePersistedCloudtentaclesContextBySourceKeys>
+
+type CloudtentaclesSourceSelectionDeps = {
+  resolveContextBySourceKeys?: typeof resolvePersistedCloudtentaclesContextBySourceKeys
+  listSourceLoadStats?: typeof listKuaishouCloudSourceLoadStats
+}
+
+export type CloudtentaclesSourceSelection = {
+  context: CloudtentaclesContext
+  sourceKey: string
+  candidateCount: number
+  activeCount: number
+  redeemingCount: number
+  inProcessCount: number
+  score: number
+  fixed: boolean
+  release: () => void
+}
+
+const inProcessSelections = new Map<string, number>()
+let selectionLockTail: Promise<void> = Promise.resolve()
+
+export async function selectCloudtentaclesSourceForFulfillment(
+  flowLike: unknown,
+  deps: CloudtentaclesSourceSelectionDeps = {},
+): Promise<CloudtentaclesSourceSelection> {
+  const flow = normalizeKuaishouCloudFlow(flowLike)
+  const resolveContext =
+    deps.resolveContextBySourceKeys || resolvePersistedCloudtentaclesContextBySourceKeys
+  const listSourceLoadStats = deps.listSourceLoadStats || listKuaishouCloudSourceLoadStats
+  const fixedSourceKey = String(flow.binding.resolvedSourceKey || '').trim()
+  const candidates = normalizeCloudtentaclesSourceCandidates(
+    fixedSourceKey
+      ? [fixedSourceKey, ...flow.binding.cloudSourceKeys]
+      : flow.binding.cloudSourceKeys,
+  )
+
+  if (fixedSourceKey) {
+    const context = resolveContext([fixedSourceKey, ...flow.binding.cloudSourceKeys])
+    return {
+      context,
+      sourceKey: context.resolvedSourceKey,
+      candidateCount: candidates.length,
+      activeCount: 0,
+      redeemingCount: 0,
+      inProcessCount: 0,
+      score: 0,
+      fixed: true,
+      release: () => {},
+    }
+  }
+
+  return withCloudtentaclesSourceSelectionLock(async () => {
+    const contexts = []
+    let lastError: unknown = null
+
+    for (const sourceKey of candidates) {
+      try {
+        contexts.push(resolveContext([sourceKey]))
+      } catch (error) {
+        lastError = error
+      }
+    }
+
+    if (contexts.length === 0) {
+      if (lastError) {
+        throw lastError
+      }
+      resolveContext(candidates)
+      throw new Error('所有 cloudtentacles 账号均不可用')
+    }
+
+    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) || {
+          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
+
+        return {
+          context,
+          sourceKey: context.resolvedSourceKey,
+          index,
+          activeCount,
+          redeemingCount,
+          inProcessCount,
+          score,
+        }
+      })
+      .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 账号均不可用')
+    }
+
+    reserveCloudtentaclesSource(selected.sourceKey)
+
+    return {
+      context: selected.context,
+      sourceKey: selected.sourceKey,
+      candidateCount: contexts.length,
+      activeCount: selected.activeCount,
+      redeemingCount: selected.redeemingCount,
+      inProcessCount: selected.inProcessCount,
+      score: selected.score,
+      fixed: false,
+      release: () => releaseCloudtentaclesSource(selected.sourceKey),
+    }
+  })
+}
+
+export function buildCloudtentaclesSourceSelectionEventDetail(
+  selection: CloudtentaclesSourceSelection,
+): JsonObject {
+  return {
+    sourceKey: selection.sourceKey,
+    candidateCount: selection.candidateCount,
+    activeCount: selection.activeCount,
+    redeemingCount: selection.redeemingCount,
+    inProcessCount: selection.inProcessCount,
+    score: selection.score,
+    fixed: selection.fixed,
+  }
+}
+
+function normalizeCloudtentaclesSourceCandidates(sourceKeys: unknown[] = []) {
+  return Array.from(new Set(normalizeStringArray(sourceKeys)))
+}
+
+async function withCloudtentaclesSourceSelectionLock<T>(fn: () => Promise<T>): Promise<T> {
+  let releaseLock: () => void = () => {}
+  const waitForTurn = selectionLockTail
+  selectionLockTail = new Promise<void>((resolve) => {
+    releaseLock = resolve
+  })
+
+  await waitForTurn
+
+  try {
+    return await fn()
+  } finally {
+    releaseLock()
+  }
+}
+
+function reserveCloudtentaclesSource(sourceKey: string) {
+  inProcessSelections.set(sourceKey, normalizeCount(inProcessSelections.get(sourceKey)) + 1)
+}
+
+function releaseCloudtentaclesSource(sourceKey: string) {
+  const nextCount = normalizeCount(inProcessSelections.get(sourceKey)) - 1
+  if (nextCount > 0) {
+    inProcessSelections.set(sourceKey, nextCount)
+    return
+  }
+
+  inProcessSelections.delete(sourceKey)
+}
+
+function normalizeCount(value: unknown) {
+  const count = Number(value || 0) || 0
+  return Number.isFinite(count) && count > 0 ? count : 0
+}