feat(aigc): WU2③ job userToken 位 + per-user 额度门(只对真实 member 生效)

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) <noreply@anthropic.com>
This commit is contained in:
lili 2026-07-07 12:22:14 -07:00
parent 4cf0e14ec3
commit a069afb649
2 changed files with 76 additions and 4 deletions

View File

@ -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<SourceProjectApi> sourceProjectApiProvider,
ObjectProvider<GenTaskProducer> genTaskProducerProvider) {
ObjectProvider<GenTaskProducer> genTaskProducerProvider,
ObjectProvider<NewapiQuotaService> 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);
}
}

View File

@ -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,<b>可为 null</b>):真实 member 在线创作派发时取其 claim 到的 token 放进
* §6.1 job 的 {@code userToken} 位;系统/编排/bake-off(无 game_player 行)旁路留空、worker 回落全局 key。
* <p><b>null 时(旧构造/单测态)</b>:不注入 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 路。