From a069afb6491e07a49ff9a15b08ea448d00d7174e Mon Sep 17 00:00:00 2001 From: lili Date: Tue, 7 Jul 2026 12:22:14 -0700 Subject: [PATCH] =?UTF-8?q?feat(aigc):=20WU2=E2=91=A2=20job=20userToken=20?= =?UTF-8?q?=E4=BD=8D=20+=20per-user=20=E9=A2=9D=E5=BA=A6=E9=97=A8=EF=BC=88?= =?UTF-8?q?=E5=8F=AA=E5=AF=B9=E7=9C=9F=E5=AE=9E=20member=20=E7=94=9F?= =?UTF-8?q?=E6=95=88=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit dispatchGeneric 组 §6.1 job 时追加 userToken 字段（向后兼容，worker .get 忽略未知键）。 per-user 门只对真实 member 在线创作生效——按 creatorUserId 是否命中一行 game_player 分流： - 无 game_player 行（系统/编排器/bake-off/admin 触发）→ 旁路：不查池不懒 claim，userToken 留空，worker 回落全局 key（不误伤 n=5 收敛环/批跑既有验证路）； - 命中 game_player 行 → 取其 CLAIMED 条目 token_key；无 CLAIMED 则派发线单 CAS 懒 claim （纯 DB CAS 非慢外呼，派发热路安全），抢到就用、池空抢不到当次干净失败 quota_exhausted。 token 随 job 内网下发日志脱敏（仅留前后 4 位）。经 12 参构造软注入，主开关关或未接线=现行行为字节零变。 Co-Authored-By: Claude Opus 4.8 (1M context) --- .../config/AigcExecutorConfiguration.java | 10 ++- .../executor/AigcGenerateExecutor.java | 70 ++++++++++++++++++- 2 files changed, 76 insertions(+), 4 deletions(-) diff --git a/game-cloud/game-module-aigc/game-module-aigc-server/src/main/java/com/wanxiang/huijing/game/module/aigc/framework/executor/config/AigcExecutorConfiguration.java b/game-cloud/game-module-aigc/game-module-aigc-server/src/main/java/com/wanxiang/huijing/game/module/aigc/framework/executor/config/AigcExecutorConfiguration.java index 79715643..b8473ccf 100644 --- a/game-cloud/game-module-aigc/game-module-aigc-server/src/main/java/com/wanxiang/huijing/game/module/aigc/framework/executor/config/AigcExecutorConfiguration.java +++ b/game-cloud/game-module-aigc/game-module-aigc-server/src/main/java/com/wanxiang/huijing/game/module/aigc/framework/executor/config/AigcExecutorConfiguration.java @@ -10,6 +10,7 @@ import com.wanxiang.huijing.game.module.aigc.service.executor.GameConfigSchemaVa import com.wanxiang.huijing.game.module.aigc.service.executor.PromptResourceLoader; import com.wanxiang.huijing.game.module.aigc.service.executor.WorkerClassifyClient; import com.wanxiang.huijing.game.module.aigc.mq.GenTaskProducer; +import com.wanxiang.huijing.game.module.aigc.service.quota.NewapiQuotaService; import com.wanxiang.huijing.game.module.aigc.saa.SaaGraphDispatcher; import com.wanxiang.huijing.game.module.aigc.service.executor.GenerationDispatcher; import com.wanxiang.huijing.game.module.aigc.service.executor.WorkerDispatchClient; @@ -178,7 +179,8 @@ public class AigcExecutorConfiguration { WorkerDispatchClient workerDispatchClient, SaaGraphDispatcher saaGraphDispatcher, ObjectProvider sourceProjectApiProvider, - ObjectProvider genTaskProducerProvider) { + ObjectProvider genTaskProducerProvider, + ObjectProvider newapiQuotaServiceProvider) { // A11 M1②：软取 SourceProjectApi（studio @Primary 就地解析；modify 路据 baseVersionId 反查 base 源注入 §6.1 job 的 sourceProject 键， // 供便宜档 HTTP worker 无 DB 被动消费——与 SaaGraphDispatcher 同一 seam）。缺席（单模块装配/studio 未在 classpath）→ modify 取源旁路（不注入 sourceProject 键），不阻断装配。 SourceProjectApi sourceProjectApi = sourceProjectApiProvider.getIfAvailable(); @@ -186,11 +188,15 @@ public class AigcExecutorConfiguration { // stale queued「重发信号」交消费者重走认领+派发，不在 tick 直派（投递可靠性补偿，让 MQ 保持唯一投递路径；与并发 cap 无关）。 // 缺席（防御分支）→ 传 null → 兜底 tick 退回「直接认领+派发」（韧性降级、不让任务卡死）。 GenTaskProducer genTaskProducer = genTaskProducerProvider.getIfAvailable(); + // WU2 §3.6：软取 NewapiQuotaService（@Service，随 aigc 模块常在席）——传入使 dispatchGeneric 组 §6.1 job 时对真实 member + // 接 per-user 额度门（取其 claim 到的 token 放进 userToken 位）；其主开关 aigc.newapi-quota.enabled 默认 false 时对所有创作者返 null， + // opt-in 才生效。缺席（防御分支/单模块装配）→ 传 null → 不接线（worker 全局 key，现行行为字节零变）。 + NewapiQuotaService newapiQuotaService = newapiQuotaServiceProvider.getIfAvailable(); // SAA 迁移 #2：传入 http worker + saa 进程内派发器，由执行器构造按 aigc.executor.dispatcher 选用 // （默认 http=现行行为不变；saa=opt-in 进程内形态①）。 return new AigcGenerateExecutor(properties, aigcTaskMapper, difyCallbackService, promptResourceLoader, schemaValidator, llmClient, workerDispatchClient, saaGraphDispatcher, - sourceProjectApi, Clock.systemDefaultZone(), genTaskProducer); + sourceProjectApi, Clock.systemDefaultZone(), genTaskProducer, newapiQuotaService); } } diff --git a/game-cloud/game-module-aigc/game-module-aigc-server/src/main/java/com/wanxiang/huijing/game/module/aigc/service/executor/AigcGenerateExecutor.java b/game-cloud/game-module-aigc/game-module-aigc-server/src/main/java/com/wanxiang/huijing/game/module/aigc/service/executor/AigcGenerateExecutor.java index aeaf9d5c..75db60f3 100644 --- a/game-cloud/game-module-aigc/game-module-aigc-server/src/main/java/com/wanxiang/huijing/game/module/aigc/service/executor/AigcGenerateExecutor.java +++ b/game-cloud/game-module-aigc/game-module-aigc-server/src/main/java/com/wanxiang/huijing/game/module/aigc/service/executor/AigcGenerateExecutor.java @@ -8,6 +8,8 @@ import com.wanxiang.huijing.game.module.aigc.mq.GenTaskProducer; import com.wanxiang.huijing.game.module.aigc.service.callback.DifyCallbackService; import com.wanxiang.huijing.game.module.studio.api.SourceProjectApi; import com.wanxiang.huijing.game.module.studio.dto.SourceProjectFetchRespDTO; +import com.wanxiang.huijing.game.module.aigc.service.quota.NewapiQuotaService; +import com.wanxiang.huijing.game.module.aigc.service.quota.QuotaPoolExhaustedException; import com.wanxiang.huijing.framework.common.enums.UserTypeEnum; import com.wanxiang.huijing.framework.common.exception.ServiceException; import com.wanxiang.huijing.framework.common.pojo.CommonResult; @@ -124,6 +126,13 @@ public class AigcGenerateExecutor { * 「扫描-认领-派发」行为，避免任务卡死（韧性降级，非主路径）。生产经 AigcExecutorConfiguration 注入非空。 */ private final GenTaskProducer genTaskProducer; + /** + * new-api per-user 额度门 seam（WU2 §3.6，可为 null）：真实 member 在线创作派发时取其 claim 到的 token 放进 + * §6.1 job 的 {@code userToken} 位；系统/编排/bake-off（无 game_player 行）旁路留空、worker 回落全局 key。 + *

null 时（旧构造/单测态）：不注入 userToken = 现行行为（worker 全局 key）。生产经 AigcExecutorConfiguration + * 软注入非空；其内部主开关 {@code aigc.newapi-quota.enabled}（默认 false）关时对所有创作者返 null，故即便在席也零行为变更、opt-in 生效。 + */ + private final NewapiQuotaService newapiQuotaService; /** 自禁用连续 tick 计数（自检恢复即清零；告警节流判据） */ private int selfCheckFailTicks = 0; @@ -200,6 +209,24 @@ public class AigcGenerateExecutor { GameConfigSchemaValidator schemaValidator, ExecutorLlmClient llmClient, WorkerDispatchClient workerDispatchClient, GenerationDispatcher saaDispatcher, SourceProjectApi sourceProjectApi, Clock clock, GenTaskProducer genTaskProducer) { + // 委托 12 参构造，newapiQuotaService 传 null → per-user 额度门不接线（现行行为：worker 全局 key，现行单测沿用此签名）。 + this(properties, aigcTaskMapper, difyCallbackService, promptResourceLoader, schemaValidator, + llmClient, workerDispatchClient, saaDispatcher, sourceProjectApi, clock, genTaskProducer, null); + } + + /** + * 12 参构造（WU2 §3.6 收口；经 AigcExecutorConfiguration @Bean 调用）：在 11 参基础上接 new-api per-user 额度门 + * {@link NewapiQuotaService}——真实 member 在线创作派发时取其 claim 到的 token 放进 §6.1 job 的 {@code userToken} 位； + * 系统/编排（无 game_player 行）旁路留空、worker 回落全局 key。可 null（旧构造/单测态）=不接线，现行行为字节零变。 + * + * @param newapiQuotaService per-user 额度门 seam（可 null=不接线；其主开关 aigc.newapi-quota.enabled 默认 false 时对所有创作者返 null，opt-in 生效） + */ + public AigcGenerateExecutor(AigcExecutorProperties properties, AigcTaskMapper aigcTaskMapper, + DifyCallbackService difyCallbackService, PromptResourceLoader promptResourceLoader, + GameConfigSchemaValidator schemaValidator, ExecutorLlmClient llmClient, + WorkerDispatchClient workerDispatchClient, GenerationDispatcher saaDispatcher, + SourceProjectApi sourceProjectApi, Clock clock, GenTaskProducer genTaskProducer, + NewapiQuotaService newapiQuotaService) { // §5.2 阈值关系铁律：stale×60 > budget + 273，装配期校验（误配置宁可启动失败也不带病收割在飞任务） properties.validateThresholds(); this.properties = properties; @@ -218,12 +245,15 @@ public class AigcGenerateExecutor { this.clock = clock; // C3：gen 队列生产者（兜底 tick 重发信号用）；非空=MQ 主触发 + 兜底重发信号（投递可靠性补偿），null=兜底退回直派（单测/降级态）。 this.genTaskProducer = genTaskProducer; + // WU2 §3.6：per-user 额度门（可 null=不接线，现行行为）。在席时其主开关关默认仍旁路，opt-in 才生效。 + this.newapiQuotaService = newapiQuotaService; // 启动三行自检日志（§12-① 冒烟观察面：执行器已启用 / prompt 资源版本 / key 配置态；严禁打印密钥本身） - log.info("[executor-selfcheck] aigc 生成执行器已启用（poll={}ms, batch={}, budget={}s, stale={}min, maxAge={}h, templates={}, sourceSeam={}, 触发={}）", + log.info("[executor-selfcheck] aigc 生成执行器已启用（poll={}ms, batch={}, budget={}s, stale={}min, maxAge={}h, templates={}, sourceSeam={}, 触发={}, perUser额度门={}）", properties.getPollIntervalMs(), properties.getScanBatchSize(), properties.getTaskBudgetSeconds(), properties.getStaleRunningMinutes(), properties.getMaxTaskAgeHours(), properties.getSupportedTemplates(), sourceProjectApi != null ? "在席" : "缺席(modify取源旁路)", - genTaskProducer != null ? "MQ主触发+tick兜底重发信号" : "tick直派(无MQ生产者/单测态)"); + genTaskProducer != null ? "MQ主触发+tick兜底重发信号" : "tick直派(无MQ生产者/单测态)", + newapiQuotaService != null ? "在席(opt-in)" : "缺席(worker全局key)"); // HJ-MC-TPL-EXEC-001 §6.4：多模板装载后无单一版本，按 templateId 逐模板列出版本（就绪时） log.info("[executor-selfcheck] prompt 资源版本 {}", promptResourceLoader.isReady() ? promptResourceLoader.describePromptVersions() : ("未就绪：" + promptResourceLoader.getNotReadyReason())); @@ -680,6 +710,29 @@ public class AigcGenerateExecutor { // 便宜档 HTTP worker（Python，无 DB）只被动消费 job["sourceProject"]（键名同 SAA 路 K_SOURCE_PROJECT）；反查失败/缺源则不放该键（best-effort 非阻断）。 putModifyFieldsIntoJob(job, task); + // ── WU2 §3.6：per-user new-api 额度门（组 §6.1 job 的 userToken 位）── + // 身份边界：只对真实 member 在线创作生效——命中一行 game_player 取其 claim 到的 token_key 放进 job； + // 系统/编排器/bake-off（无 game_player 行）旁路留空、worker 回落全局 key（§3.7），不误伤既有验证路。 + // newapiQuotaService 为 null（旧构造/单测态）或主开关关 → resolveUserTokenForDispatch 返 null，不注入 = 现行行为。 + if (newapiQuotaService != null) { + try { + String userToken = newapiQuotaService.resolveUserTokenForDispatch(task.getCreatorUserId()); + if (StringUtils.hasText(userToken)) { + job.put("userToken", userToken); // §6.1 追加字段（向后兼容：worker .get 忽略未知键，老 worker 收到不炸） + // 传输安全（§3.8）：token 是 new-api 调用凭据，仅留前后 4 位入日志，绝不整条打印。 + log.info("[executor-dispatch] per-user 额度 token 已随 job 下发（脱敏）taskId={}, traceId={}, creatorUserId={}, tokenMasked={}", + task.getId(), task.getTraceId(), task.getCreatorUserId(), maskToken(userToken)); + } + // userToken 为空 = 系统/编排旁路（正常且预期），不注入、不打扰日志。 + } catch (QuotaPoolExhaustedException e) { + // 真实 member 需要 token 但池空、懒 claim 抢不到 → 当次干净失败给可读拒因（§3.6 池空处置）； + // 绝不派一个没 token 的 create 路 job 下去静默烧共享额度。水位告警已在 service 内触发。 + log.warn("[executor-dispatch] per-user 额度池空/懒 claim 未果，按 quota_exhausted 干净失败 taskId={}, traceId={}, creatorUserId={}", + task.getId(), task.getTraceId(), task.getCreatorUserId()); + return callbackFailed(task, FailureReasonEnum.QUOTA_EXHAUSTED); + } + } + boolean dispatched = generationDispatcher.dispatch(job); if (dispatched) { // 投递成功：任务留 RUNNING 等 worker 回调（终态由回调驱动，不在此写）。tick 汇总以「派发」计数（非 succeeded/failed）。 @@ -694,6 +747,19 @@ public class AigcGenerateExecutor { return callbackFailed(task, FailureReasonEnum.LLM_ERROR); } + /** + * per-user token 日志脱敏（WU2 §3.8）：只留前后各 4 位、中间打码，绝不整条打印 new-api 调用凭据。 + * + * @param token 原始 token_key（可能为空） + * @return 脱敏串（如 sk-A****Z9zK）；空/过短返回固定占位 + */ + private static String maskToken(String token) { + if (token == null || token.length() <= 8) { + return "****"; + } + return token.substring(0, 4) + "****" + token.substring(token.length() - 4); + } + /** * 把 modify/extend 四件套（契约 C3）接进 §6.1 job 的 modify 区——使 {@link com.wanxiang.huijing.game.module.aigc.saa.SaaGraphDispatcher} * 的 {@code buildInputs} 能读到 modifyMode/modifyPatch/baseVersionId 并据此分流到 SAA modify 路。