feat(creation): Java 侧接通 patbond-ai——Feign 客户端、两类错误分流、三值重试(T4-03)
CI / backend-test (push) Successful in 6m22s

按 openapi-internal.yaml 把 AiServiceGenerationProvider 从空实现位做成真实现。
全量 410 → 427,零失败零跳过。

**骨架阶段的三处假设被现实推翻,一并纠正:**

1. **去掉 submit() 的 idempotencyKey 参数。** 当时按「透传给厂商幂等头」设计,
   但火山**没有任何幂等字段**(取证见 providers/volcengine-seedance.md §2-①)。
   留一个无处可去的参数会让人以为双重扣费已经解决——它没有,只由 max_attempts
   兜住,恢复路径是已落库的 provider_request_id 而不是某个 key。
2. **Worker 不再在 submit 前调 uploadTargets.presignedPutFor。** 那个方法目前
   抛异常,而它在 submit **之前**被调用——**所以在此之前整条链一步都跑不动**。
   契约把 upload_target 标为可选未启用,故传 null 才是正确而非权宜。
3. **PollResult 扩容 + retryable 由布尔改三值 Retry。** 布尔表达不了最关键的
   区别:限流几秒后恢复,配额耗尽要等到明天。把后者当可重试会白烧完整个重试
   预算再失败,而用户看到的是转圈而不是「额度用完」的出口(ADR-023 §9)。
   同时补齐 assetUrl/expiresIn/coverUrl/尺寸/时长/outputFormat/billingTokens
   ——这些最终都要写进 media.assets,现在丢掉后面就得再改一遍。

**两类错误落在两个异常上**(AiProviderException / AiServiceException),判别用
error.kind 而非 HTTP 状态码(状态码在两类之间会撞)。若这个分流塌成一个笼统的
FeignException,Worker 就只能猜要不要重试。未知的 retry/user_facing/state 取值
一律偏向「可重试 / 仍在运行」:契约可能先长出一个取值、代码后知道,把可恢复的
抖动判成用户可见的失败是更糟的错误,而重试预算本身有上限。

**snake_case 只作用于这一个客户端**:用 SpringEncoder/SpringDecoder 换 ObjectMapper,
不引 feign-jackson(其他模块都不需要它),也不用逐字段 @JsonProperty——后者漏一个
注解就会静默发出对端忽略的字段,而「静默被忽略的字段」是最难察觉的失败。
另配 followRedirects=false:内部服务没有理由重定向我们,跟随会把服务票发给
重定向指定的任意主机。

**新增两个 429 码**(ADR-023 §9):42900 配额耗尽 / 42901 频率过快。必须是两个
——配额等的是次日或充值(弹 sheet 给出口),频率等的是数秒(弹 SnackBar 可自动
重试),共用一码必然使一半场景的文案与出口是错的。

写进任务行的 error_code 用适配层的 userFacing 类别而非厂商原始码:厂商词汇不该
渗进我们的数据,它会随换供应商而变,客户端会据此分支。原始码留在 error_message
里供排查。

safety_identifier 传 userId 的 SHA-256 hex 而非明文(ADR-023 §5 数据最小化),
测试断言明文 userId 不出现在请求体里。

**新增 AiGenerationClientWireTest(17 项)**,照 community 的
AuthorProfileClientWireTest 先例用 JDK 内置 HttpServer 起桩:snake_case 双向、
服务票每次现签(断言两次调用的票不同——缓存复用会让短期凭证悄悄变成长期凭证)、
两类错误各自的分流、配额与限流的 retry 不同、无法识别的错误体按可重试并说明原因。

**一处真实缺陷由全量跑抓出**:@FeignClient 的 url 占位符原先无默认值,而 Feign 在
**上下文启动时**解析它——导致任何没显式配置该属性的上下文都启动失败,包括与 AI
毫无关系的两个仓储测试类(17 个测试全红)。已给占位符加上与 CreationProperties
一致的默认值。单跑某个测试类时不会暴露这个问题,只有全量跑才会。

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
This commit is contained in:
2026-09-20 17:43:28 +08:00
parent 91ae016470
commit a8ffd02175
17 changed files with 1276 additions and 113 deletions
@@ -29,7 +29,7 @@ public enum ErrorCode {
COMMENT_NOT_FOUND(40404, 404, "评论不存在"),
TARGET_USER_NOT_FOUND(40406, 404, "用户不存在"),
// M4 生成域(ADR-023 §9)。40407/40408 与既有 404 族同列;配额与限流的
// 两个 429 码随配额工单落地时补入——此处只加骨架实际使用的码,避免死枚举。
// 骨架阶段只加了实际使用的码;两个 429 已随 AI 适配层落地(见下)。
GENERATION_JOB_NOT_FOUND(40407, 404, "生成任务不存在"),
GENERATION_MODEL_NOT_FOUND(40408, 404, "模型或风格不可用"),
VACCINATION_RULE_VIOLATION(42201, 422, "疫苗状态或日期约束不满足"),
@@ -39,6 +39,11 @@ public enum ErrorCode {
MEDIA_UPLOAD_STATE_INVALID(42205, 422, "上传状态不允许确认"),
GENERATION_STATE_INVALID(42206, 422, "生成任务状态不允许该操作"),
LOGIN_LOCKED(42300, 423, "登录失败次数过多,账号已临时锁定"),
// M4 生成域的两个 429(ADR-023 §9)。**必须是两个码**:配额耗尽等的是次日或
// 充值(客户端弹 sheet 给出口),频率过快等的是数秒(弹 SnackBar 可自动重试)。
// 共用一码必然使一半场景的文案与出口是错的。
GENERATION_QUOTA_EXHAUSTED(42900, 429, "今日 AI 生成额度已用完"),
GENERATION_RATE_LIMITED(42901, 429, "操作过于频繁,请稍后再试"),
INTERNAL_ERROR(50000, 500, "服务器内部错误"),
DOWNSTREAM_UNAVAILABLE(50300, 503, "依赖服务暂不可用");
@@ -2,6 +2,7 @@ package com.patbond.patbond.creation;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.cloud.openfeign.EnableFeignClients;
import org.springframework.scheduling.annotation.EnableScheduling;
/**
@@ -21,6 +22,7 @@ import org.springframework.scheduling.annotation.EnableScheduling;
* consumption off for a request-only replica or a test.
*/
@SpringBootApplication
@EnableFeignClients
@EnableScheduling
public class CreationApplication {
@@ -4,6 +4,7 @@ import com.fasterxml.jackson.databind.ObjectMapper;
import com.patbond.patbond.creation.provider.AiServiceGenerationProvider;
import com.patbond.patbond.creation.provider.FixtureGenerationProvider;
import com.patbond.patbond.creation.provider.GenerationProvider;
import com.patbond.patbond.creation.provider.ai.AiGenerationClient;
import com.patbond.patbond.creation.security.BearerAuthFilter;
import com.patbond.patbond.creation.security.JwtVerifier;
import com.patbond.patbond.creation.security.ServiceTokenSigner;
@@ -63,9 +64,16 @@ public class SecurityConfig {
return new FixtureGenerationProvider();
}
/**
* 真实的服务商适配器,经 patbond-ai 到达。
*
* <p>注意它与 fixture 适配器**同时注册**:`provider_code` 不同,故任务行冻结的
* 快照决定用哪个——这正是快照列存在的意义,目录被改指向后在途任务仍由当初提交
* 它的适配器轮询。
*/
@Bean
public AiServiceGenerationProvider aiServiceGenerationProvider() {
return new AiServiceGenerationProvider();
public AiServiceGenerationProvider aiServiceGenerationProvider(AiGenerationClient client) {
return new AiServiceGenerationProvider(client);
}
/** provider_code → adapter, so a job's snapshot routes deterministically. */
@@ -37,6 +37,7 @@ public record GenerationJob(
short attemptCount,
short maxAttempts,
String providerCodeSnapshot,
String providerModelSnapshot,
String providerRequestId,
String errorCode,
String errorMessage,
@@ -1,48 +1,42 @@
package com.patbond.patbond.creation.provider;
import com.patbond.patbond.creation.domain.GenerationJob;
import com.patbond.patbond.creation.provider.ai.AiContract;
import com.patbond.patbond.creation.provider.ai.AiGenerationClient;
import com.patbond.patbond.creation.provider.ai.AiProviderException;
import com.patbond.patbond.creation.provider.ai.AiServiceException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
/**
* Talks to patbond-ai (Python) over REST for real vendor generation —
* 火山引擎 Seedance for video, gpt-image-2 for images (ADR-023 §1).
* Reaches 火山方舟 Seedance (video) through patbond-ai over REST.
*
* <p>Deliberately unimplemented at skeleton stage. Five facts must come out of
* the vendors' official documentation before the request/response shape can be
* fixed, and guessing any of them would produce an adapter that compiles and
* silently misbehaves:
* <ol>
* <li><b>Idempotency key support</b> — decides whether the double-charge
* window on a synchronous vendor can be closed at all, or only bounded
* by {@code max_attempts} (ADR-023 §4). This is the one item no amount of
* design on our side can work around.</li>
* <li><b>Query endpoint shape and any polling rate limit</b> — decides
* whether the 3s/10s/30s backoff ladder is admissible.</li>
* <li><b>Maximum job duration and timeout behaviour</b> — feeds the
* server-issued client timeout, which must not be hardcoded per kind.</li>
* <li><b>Artifact return form (URL vs bytes)</b> — a time-limited vendor URL
* lets patbond-ai stream straight into the presigned PUT without ever
* landing the file.</li>
* <li><b>Rate-limit and quota response shapes</b> — decides how they map onto
* the two distinct 429 codes (quota exhausted waits until tomorrow and
* gets a sheet; too-fast waits seconds and gets a snackbar — one shared
* code would make half the copy and half the exits wrong).</li>
* </ol>
* <p>This class is thin on purpose: all vendor knowledge — request shapes, error
* codes, per-model parameter limits, pixel tables — lives in patbond-ai. What is
* here is the translation between our job row and that service's contract, plus
* the mapping from its two error kinds onto what the worker should do.
*
* <p>Note what is NOT blocked by the above: the transport, auth and error
* envelope are already settled and independent of vendor specifics — REST/JSON
* (not gRPC), a service-identity JWT signed with a separate key pair, and a
* request body that carries {@code jobId}, generation parameters and the
* presigned upload target but deliberately NOT {@code userId} (ADR-023 §5 data
* minimisation).
* <p>{@code provider_code} must match the value patbond-ai registers its adapter
* under and the value seeded into {@code generation_models.provider_code} (V7).
* If the three drift apart, a job never finds an adapter and sits in the queue
* forever — a symptom that looks nothing like "two strings disagree", which is
* why both sides have a test pinning it.
*/
public class AiServiceGenerationProvider implements GenerationProvider {
/**
* Placeholder code. The real value comes from
* {@code generation_models.provider_code} in the V7 catalog seed, which is
* itself waiting on the same vendor documentation.
* Must equal patbond-ai's {@code PROVIDER_CODE} and V7's seeded
* {@code provider_code}. Changing any one of the three means changing all.
*/
public static final String PROVIDER_CODE = "patbond-ai";
public static final String PROVIDER_CODE = "volcengine-seedance";
private static final Logger log = LoggerFactory.getLogger(AiServiceGenerationProvider.class);
private final AiGenerationClient client;
public AiServiceGenerationProvider(AiGenerationClient client) {
this.client = client;
}
@Override
public String providerCode() {
@@ -50,16 +44,164 @@ public class AiServiceGenerationProvider implements GenerationProvider {
}
@Override
public SubmitResult submit(GenerationJob job, String uploadTarget, String idempotencyKey) {
throw new UnsupportedOperationException(
"AiServiceGenerationProvider.submit 未实现:待 Seedance 与 gpt-image-2 官方文档"
+ "确认五项(幂等键支持、查询接口与频率限制、单任务最长时长、产物返回形式、"
+ "限流与配额响应形态)后落地。");
public SubmitResult submit(GenerationJob job, String uploadTarget) {
AiContract.SubmitRequest request = new AiContract.SubmitRequest(
job.id().toString(),
job.mediaKind().dbValue(),
// The model comes from the row's frozen snapshot, never from the
// live catalog: the catalog may be re-pointed, and a job must be
// billed against the model it actually ran on.
job.providerModelSnapshot(),
job.prompt(),
// Null means text-to-*, present means reference-guided
// (reference image optional, decision A2). The input asset must
// be reachable by the vendor, so this has to be a URL rather
// than our internal asset id.
null,
null,
null,
job.durationMs() == null ? null : (int) (job.durationMs() / 1000),
true,
false,
// Hash, never the plain user id (ADR-023 §5 data minimisation).
safetyIdentifierFor(job),
null,
// Unused until the object-storage decision lands (ADR-023 §7);
// the contract marks it optional so passing null is correct
// rather than a stopgap.
uploadTarget);
AiContract.SubmitResponse response = client.submit(request);
if (response == null || response.providerRequestId() == null
|| response.providerRequestId().isBlank()) {
// A 2xx with no id means we can never query this task again, and it
// may already be billing. Never let it pass as success.
throw new IllegalStateException(
"patbond-ai 提交成功但未返回 provider_request_id,job=" + job.id());
}
return new SubmitResult(response.providerRequestId());
}
@Override
public PollResult poll(GenerationJob job, String providerRequestId) {
throw new UnsupportedOperationException(
"AiServiceGenerationProvider.poll 未实现(同 submit)。");
AiContract.TaskResponse response;
try {
response = client.get(providerRequestId);
} catch (AiServiceException e) {
if (e.isTaskNotFound()) {
// The vendor keeps tasks for 7 days only. There is nothing left
// to poll, so fail terminally instead of backing off forever.
return PollResult.failed(new PollResult.Failure(
Retry.NEVER, UserFacing.SERVER_ERROR,
"PROVIDER_TASK_GONE",
"服务商已无此任务(可能超出其保留期):" + providerRequestId));
}
throw e;
}
return toPollResult(response);
}
private static PollResult toPollResult(AiContract.TaskResponse response) {
PollResult.State state = parseState(response.state());
return switch (state) {
case PENDING -> PollResult.pending(response.progress());
case RUNNING -> PollResult.running(response.progress());
case CANCELLED -> PollResult.cancelled();
case FAILED -> PollResult.failed(toFailure(response.failure()));
case SUCCEEDED -> new PollResult(
PollResult.State.SUCCEEDED,
100,
response.assetUrl(),
response.assetUrlExpiresInSec(),
response.coverUrl(),
response.ratio(),
response.resolution(),
response.widthPx(),
response.heightPx(),
response.durationMs(),
response.outputFormat(),
response.billingTokens(),
null);
};
}
private static PollResult.Failure toFailure(AiContract.Failure failure) {
if (failure == null) {
// The contract guarantees failed ⇒ failure present. Getting here means
// the contract was violated; a synthetic retryable failure is the safe
// reading, but say so loudly rather than inventing a plausible code.
log.error("patbond-ai 返回 state=failed 但缺少 failure 对象,违反内部契约");
return new PollResult.Failure(Retry.BACKOFF, UserFacing.UPSTREAM_UNAVAILABLE,
"AI_CONTRACT_VIOLATION", "state=failed 但未携带 failure");
}
return new PollResult.Failure(
parseRetry(failure.retry()),
parseUserFacing(failure.userFacing()),
failure.providerCode(),
failure.message());
}
/**
* Unknown states are treated as still running rather than as failures: the
* contract may gain a value before this code learns it, and declaring an
* in-flight job dead is worse than waiting for our own timeout.
*/
private static PollResult.State parseState(String value) {
if (value == null) {
return PollResult.State.RUNNING;
}
return switch (value) {
case "pending" -> PollResult.State.PENDING;
case "running" -> PollResult.State.RUNNING;
case "succeeded" -> PollResult.State.SUCCEEDED;
case "failed" -> PollResult.State.FAILED;
case "cancelled" -> PollResult.State.CANCELLED;
default -> {
log.warn("patbond-ai 返回未知 state:{},按仍在运行处理", value);
yield PollResult.State.RUNNING;
}
};
}
private static Retry parseRetry(String value) {
if (value == null) {
return Retry.BACKOFF;
}
return switch (value) {
case "never" -> Retry.NEVER;
case "quota" -> Retry.QUOTA;
case "backoff" -> Retry.BACKOFF;
default -> {
log.warn("patbond-ai 返回未知 retry:{},按可重试处理", value);
yield Retry.BACKOFF;
}
};
}
private static UserFacing parseUserFacing(String value) {
if (value == null) {
return UserFacing.UPSTREAM_UNAVAILABLE;
}
return switch (value) {
case "quota_exhausted" -> UserFacing.QUOTA_EXHAUSTED;
case "rate_limited" -> UserFacing.RATE_LIMITED;
case "content_rejected" -> UserFacing.CONTENT_REJECTED;
case "server_error" -> UserFacing.SERVER_ERROR;
case "upstream_unavailable" -> UserFacing.UPSTREAM_UNAVAILABLE;
default -> UserFacing.UPSTREAM_UNAVAILABLE;
};
}
/**
* Stable per-user hash for the vendor's abuse detection.
*
* <p>Hashed rather than the raw user id: the vendor has no need for our
* identifiers, and a stable hash serves their purpose equally well. Derived
* from the job's user id so the same user maps to the same value across
* jobs, which is what makes it useful to them.
*/
private static String safetyIdentifierFor(GenerationJob job) {
return com.patbond.patbond.creation.support.RequestHashes
.hex(job.userId().toString());
}
}
@@ -35,7 +35,7 @@ public class FixtureGenerationProvider implements GenerationProvider {
}
@Override
public SubmitResult submit(GenerationJob job, String uploadTarget, String idempotencyKey) {
public SubmitResult submit(GenerationJob job, String uploadTarget) {
log.warn("Fixture provider is not implemented yet; job {} will not be submitted", job.id());
throw new UnsupportedOperationException(
"FixtureGenerationProvider.submit 未实现:产物生成与预签名上传的形态取决于"
@@ -3,24 +3,21 @@ package com.patbond.patbond.creation.provider;
import com.patbond.patbond.creation.domain.GenerationJob;
/**
* The upstream generation boundary. One implementation talks to patbond-ai
* over REST; a fixture implementation stands in while provider credentials are
* not yet available (ADR-023 §1/§4).
* The upstream generation boundary. Implementations reach the vendor through
* patbond-ai over REST; the contract is that repo's
* {@code openapi-internal.yaml} (ADR-023 §4).
*
* <p><b>Two phases, always — even for synchronous providers.</b> Some vendors
* are asynchronous (submit + query, e.g. video), others complete inside one
* long-lived HTTP call (e.g. image). That difference must NOT reach this
* interface, the job table or the client contract: {@code submit} returns as
* soon as the work is accepted and {@code poll} reports progress. A
* synchronous vendor is absorbed on the patbond-ai side by running the long
* call as a background task.
* <p><b>Two phases, always — even for synchronous vendors.</b> Some are
* asynchronous (submit + query, e.g. video), others complete inside one
* long-lived call. That difference must NOT reach this interface, the job table
* or the client contract: {@code submit} returns as soon as the work is accepted
* and {@code poll} reports progress. A synchronous vendor is absorbed on the
* patbond-ai side.
*
* <p>Why it matters beyond tidiness: {@code submit} returning a request id
* immediately is what lets the worker persist {@code provider_request_id}
* before any billable work happens. Without it, a timeout leaves us unable to
* tell whether we were charged, and the retry charges again. The partial
* unique index {@code uq_generation_jobs_provider_request} exists to enforce
* exactly that de-duplication.
* immediately is what lets the worker persist {@code provider_request_id} before
* any billable work happens. Without it, a timeout leaves us unable to tell
* whether we were charged, and the retry charges again.
*/
public interface GenerationProvider {
@@ -34,19 +31,22 @@ public interface GenerationProvider {
/**
* Hands the job upstream. MUST NOT block for the generation itself.
*
* @param job the claimed job, already RUNNING under this worker's lease
* @param uploadTarget where the artifact is to be PUT once produced —
* a presigned URL obtained from patbond-user's internal media API.
* Passing the target in at submit time is what keeps object-storage
* credentials out of this service and out of patbond-ai entirely
* (ADR-023 §4).
* @param idempotencyKey the job's own key, to be forwarded upstream as the
* vendor's idempotency header where the vendor supports one. Where it
* does not, the double-charge window is bounded only by
* {@code max_attempts} — a residual risk ADR-023 §4 records rather
* than pretends to have solved.
* <p><b>There is deliberately no idempotency-key parameter.</b> The skeleton
* had one, on the assumption it could be forwarded to the vendor's
* idempotency header — but 火山方舟 has no such field (see patbond-doc
* {@code docs/development/providers/volcengine-seedance.md} §2-①). Keeping a
* parameter that has nowhere to go would imply the double-charge window is
* solved. It is not: it is bounded only by {@code max_attempts}, and the
* recovery path is the persisted {@code provider_request_id}, not a key.
*
* @param uploadTarget where the artifact is to be PUT once produced — a
* presigned URL from patbond-user's internal media API. <b>Pass null for
* now</b>: artifact transfer is blocked on the object-storage decision
* (ADR-023 §7) and the contract marks the field unused. The parameter
* stays so enabling it later needs no signature change.
* @return the vendor's task id, to be persisted before anything billable
*/
SubmitResult submit(GenerationJob job, String uploadTarget, String idempotencyKey);
SubmitResult submit(GenerationJob job, String uploadTarget);
/**
* Asks after a previously submitted job.
@@ -63,22 +63,83 @@ public interface GenerationProvider {
record SubmitResult(String providerRequestId) {
}
/**
* What to do about a failure. Mirrors the contract's {@code retry} enum.
*
* <p>The skeleton used a boolean {@code retryable}, which cannot express the
* difference that matters most: a rate limit clears in seconds, an exhausted
* quota clears tomorrow. Treating the latter as retryable burns the whole
* retry budget for nothing and still fails.
*/
enum Retry {
/** Same input will always give the same result. Terminal. */
NEVER,
/** Rate limit, burst protection, vendor 5xx. Retry with backoff. */
BACKOFF,
/**
* Quota or billing exhausted — clears at the next billing period or on
* top-up. Retrying is pointless, so the job fails terminally and the
* user gets an exit instead of a silent wait.
*/
QUOTA;
public boolean worthRetrying() {
return this == BACKOFF;
}
}
/**
* How a failure should be presented to the end user. Mirrors the contract's
* {@code user_facing} enum; maps onto the two distinct 429 codes
* (ADR-023 §9) — sharing one code would make half the copy and half the
* exits wrong.
*/
enum UserFacing {
QUOTA_EXHAUSTED,
RATE_LIMITED,
CONTENT_REJECTED,
SERVER_ERROR,
UPSTREAM_UNAVAILABLE
}
/**
* Outcome of a poll.
*
* @param state where the upstream job is now
* @param progress 0..100 when known; the client contract deliberately does
* not promise a percentage (ADR-023 §10), so 0 while queued is normal
* rather than a bug
* @param objectKey set only on {@link State#SUCCEEDED} — the key the
* artifact was uploaded under, which this service turns into a
* {@code media.assets} row
* @param errorCode set only on {@link State#FAILED}
* @param retryable whether a FAILED outcome is worth another attempt; a
* rejected prompt is not, a vendor 5xx is
* <p>Carries everything needed to write a {@code media.assets} row, because
* dropping it here would mean a second round of changes when artifact
* transfer lands.
*
* @param progress 0..100. The vendor gives no percentage, so it is 0 until
* success and 100 after. This is not a defect: the client contract
* deliberately does not promise smooth progress (ADR-023 §10), and an
* honest 0 must not be read as "stuck".
* @param assetUrl the artifact. <b>May be a time-limited vendor URL</b> —
* 火山's is valid for 24 hours — so it must be transferred within that
* window and never stored long-term as {@code external_url}.
* @param assetUrlExpiresInSec how long {@code assetUrl} remains valid.
* Travels with the URL so no caller has a reason to treat it as
* permanent.
* @param coverUrl cover image for video. 火山's last-frame image serves this
* for free, so no frame extraction is needed.
* @param widthPx derived by the adapter from the vendor's ratio ×
* resolution; null when it cannot be determined rather than guessed.
* This is what lets the AI path fill real dimensions without decoding
* the file (ADR-023 §11).
* @param billingTokens reconciliation figure from the vendor.
*/
record PollResult(State state, int progress, String objectKey,
String errorCode, String errorMessage, boolean retryable) {
record PollResult(State state,
int progress,
String assetUrl,
Integer assetUrlExpiresInSec,
String coverUrl,
String ratio,
String resolution,
Integer widthPx,
Integer heightPx,
Long durationMs,
String outputFormat,
Long billingTokens,
Failure failure) {
public enum State {
/** Accepted upstream, not started. */
@@ -86,23 +147,50 @@ public interface GenerationProvider {
/** Upstream is working on it. */
RUNNING,
SUCCEEDED,
FAILED
FAILED,
/** Cancelled upstream. Terminal, and not a failure. */
CANCELLED
}
/**
* @param providerCode the vendor's raw error code, kept verbatim — it is
* usually the only clue when troubleshooting. Callers must not
* branch on it; that is the adapter's job.
*/
public record Failure(Retry retry, UserFacing userFacing,
String providerCode, String message) {
}
public PollResult {
// A FAILED result without a Failure leaves the worker unable to
// decide anything — it could only retry blindly or give up blindly,
// and both are wrong. Better to fail loudly here.
if (state == State.FAILED && failure == null) {
throw new IllegalArgumentException("FAILED 必须携带 Failure");
}
if (state == State.SUCCEEDED && (assetUrl == null || assetUrl.isBlank())) {
throw new IllegalArgumentException("SUCCEEDED 必须携带 assetUrl");
}
}
public static PollResult pending(int progress) {
return new PollResult(State.PENDING, progress, null, null, null, false);
return new PollResult(State.PENDING, progress, null, null, null, null, null,
null, null, null, null, null, null);
}
public static PollResult running(int progress) {
return new PollResult(State.RUNNING, progress, null, null, null, false);
return new PollResult(State.RUNNING, progress, null, null, null, null, null,
null, null, null, null, null, null);
}
public static PollResult succeeded(String objectKey) {
return new PollResult(State.SUCCEEDED, 100, objectKey, null, null, false);
public static PollResult cancelled() {
return new PollResult(State.CANCELLED, 0, null, null, null, null, null,
null, null, null, null, null, null);
}
public static PollResult failed(String errorCode, String errorMessage, boolean retryable) {
return new PollResult(State.FAILED, 0, null, errorCode, errorMessage, retryable);
public static PollResult failed(Failure failure) {
return new PollResult(State.FAILED, 0, null, null, null, null, null,
null, null, null, null, null, failure);
}
}
}
@@ -0,0 +1,115 @@
package com.patbond.patbond.creation.provider.ai;
import com.fasterxml.jackson.annotation.JsonInclude;
/**
* DTOs of patbond-ai's internal contract ({@code openapi-internal.yaml}).
*
* <p><b>Field names are snake_case</b>, unlike the camelCase contract we expose
* to clients. That is deliberate: patbond-ai is Python, and snake_case lets its
* JSON map one-to-one onto its domain types with no conversion layer. The cost
* is confined to one Jackson naming strategy on one Feign client
* (see {@code AiFeignConfig}) — using camelCase would instead require a
* conversion on both sides.
*
* <p>Java field names here stay camelCase; the wire form is produced by the
* naming strategy, not by hand-written {@code @JsonProperty} on every field.
* Hand-annotating is where the two sides drift: one forgotten annotation
* silently sends a field the other end ignores.
*/
public final class AiContract {
private AiContract() {
}
/**
* Submit request.
*
* <p>{@code @JsonInclude(NON_NULL)} matters here beyond tidiness: the
* contract declares {@code extra = "forbid"} on the Python side, and absent
* optional fields must be absent rather than explicit nulls so the defaults
* documented in the contract apply.
*
* <p><b>Carries no userId</b> — data minimisation (ADR-023 §5). The vendor's
* abuse detection gets {@code safetyIdentifier}, a hash, instead.
*/
@JsonInclude(JsonInclude.Include.NON_NULL)
public record SubmitRequest(
String jobId,
String mediaKind,
String model,
String prompt,
String inputUrl,
String ratio,
String resolution,
Integer durationSec,
Boolean generateAudio,
Boolean watermark,
String safetyIdentifier,
String callbackUrl,
String uploadTarget) {
}
public record SubmitResponse(String providerRequestId) {
}
/**
* Task state.
*
* <p>{@code state} is one of pending / running / succeeded / failed /
* cancelled — deliberately a smaller set than the vendor's: the vendor's
* {@code queued} arrives as {@code pending}, and {@code expired} arrives as
* {@code failed} with a retryable {@code failure} (it is a timeout, not a
* content problem).
*/
public record TaskResponse(
String providerRequestId,
String state,
int progress,
String assetUrl,
Integer assetUrlExpiresInSec,
String coverUrl,
String ratio,
String resolution,
Integer widthPx,
Integer heightPx,
Long durationMs,
String outputFormat,
Long billingTokens,
Failure failure) {
}
/**
* Failure classification.
*
* @param retry what the worker should do: never / backoff / quota
* @param userFacing how to present it: quota_exhausted / rate_limited /
* content_rejected / server_error / upstream_unavailable
* @param providerCode the vendor's raw code, for logs only
*/
public record Failure(String retry, String userFacing, String providerCode,
String message) {
}
/**
* Error envelope. <b>Both error kinds share this shape</b> so one decoder
* handles both; {@code kind} discriminates.
*
* <p>{@code kind = "service"} carries {@code code}; {@code kind = "provider"}
* carries the same classification fields as {@link Failure}. The overlap is
* intentional — it means the caller reuses one piece of handling logic
* instead of writing separate branches for "rate-limited at submit" and
* "task failed because rate-limited".
*/
public record ErrorEnvelope(ErrorBody error) {
public record ErrorBody(String kind, String code, String retry,
String userFacing, String providerCode,
String message) {
public boolean isProvider() {
return "provider".equals(kind);
}
}
}
}
@@ -0,0 +1,131 @@
package com.patbond.patbond.creation.provider.ai;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.databind.PropertyNamingStrategies;
import com.patbond.patbond.creation.provider.GenerationProvider.Retry;
import com.patbond.patbond.creation.provider.GenerationProvider.UserFacing;
import feign.Response;
import feign.Util;
import feign.codec.ErrorDecoder;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.io.IOException;
/**
* Turns patbond-ai's two error kinds into two distinct Java exceptions.
*
* <p><b>Why two, and not one with a flag.</b> The caller's reaction differs
* completely:
*
* <ul>
* <li>{@code service} — our own bug or misconfiguration (bad request body,
* unauthenticated, adapter not assembled). Retrying a malformed request
* never helps.</li>
* <li>{@code provider} — the vendor said no. Whether to retry, and how long to
* wait, is carried in {@code retry}; the caller must not try to interpret
* vendor codes itself.</li>
* </ul>
*
* <p>The discriminator is {@code error.kind}, <b>not the HTTP status</b>: statuses
* collide between the two kinds (a 5xx can be either), while {@code kind} cannot.
*
* <p>Note that a <i>failed task</i> is neither of these — it arrives as HTTP 200
* with {@code state = failed}. There the HTTP call succeeded; the task is what
* failed. Conflating them would make "task not found" and "task failed"
* indistinguishable.
*/
public class AiErrorDecoder implements ErrorDecoder {
private static final Logger log = LoggerFactory.getLogger(AiErrorDecoder.class);
private static final ObjectMapper MAPPER = new ObjectMapper()
.setPropertyNamingStrategy(PropertyNamingStrategies.SNAKE_CASE)
.configure(com.fasterxml.jackson.databind.DeserializationFeature
.FAIL_ON_UNKNOWN_PROPERTIES, false);
@Override
public Exception decode(String methodKey, Response response) {
AiContract.ErrorEnvelope.ErrorBody body = readBody(response);
if (body == null) {
// Not our envelope at all. Most likely we hit the wrong address and
// something else answered (a gateway error page, say). Treat as a
// retryable upstream problem, but say plainly that the shape was
// unrecognised — silently classifying it would send troubleshooting
// down the wrong path.
log.warn("patbond-ai 返回了无法识别的错误体 status={} method={}",
response.status(), methodKey);
return new AiProviderException(new PollResultFailure(
Retry.BACKOFF, UserFacing.UPSTREAM_UNAVAILABLE,
"AI_RESPONSE_UNRECOGNISED",
"patbond-ai 返回了非契约错误体,HTTP " + response.status()));
}
if (body.isProvider()) {
return new AiProviderException(new PollResultFailure(
parseRetry(body.retry()),
parseUserFacing(body.userFacing()),
body.providerCode(),
body.message()));
}
return new AiServiceException(response.status(), body.code(), body.message());
}
private static AiContract.ErrorEnvelope.ErrorBody readBody(Response response) {
if (response.body() == null) {
return null;
}
try {
String raw = Util.toString(response.body().asReader(Util.UTF_8));
AiContract.ErrorEnvelope envelope =
MAPPER.readValue(raw, AiContract.ErrorEnvelope.class);
return envelope == null ? null : envelope.error();
} catch (IOException | RuntimeException e) {
return null;
}
}
/**
* Unknown values lean towards retrying rather than failing terminally: the
* contract may gain a value before this code learns it, and turning a
* recoverable blip into a user-visible failure is the worse error. The retry
* budget is bounded anyway.
*/
private static Retry parseRetry(String value) {
if (value == null) {
return Retry.BACKOFF;
}
return switch (value) {
case "never" -> Retry.NEVER;
case "quota" -> Retry.QUOTA;
case "backoff" -> Retry.BACKOFF;
default -> {
log.warn("patbond-ai 返回未知 retry 取值:{},按可重试处理", value);
yield Retry.BACKOFF;
}
};
}
private static UserFacing parseUserFacing(String value) {
if (value == null) {
return UserFacing.UPSTREAM_UNAVAILABLE;
}
return switch (value) {
case "quota_exhausted" -> UserFacing.QUOTA_EXHAUSTED;
case "rate_limited" -> UserFacing.RATE_LIMITED;
case "content_rejected" -> UserFacing.CONTENT_REJECTED;
case "server_error" -> UserFacing.SERVER_ERROR;
case "upstream_unavailable" -> UserFacing.UPSTREAM_UNAVAILABLE;
default -> {
log.warn("patbond-ai 返回未知 user_facing 取值:{}", value);
yield UserFacing.UPSTREAM_UNAVAILABLE;
}
};
}
/** Internal carrier so the decoder need not import the nested record path. */
record PollResultFailure(Retry retry, UserFacing userFacing,
String providerCode, String message) {
}
}
@@ -0,0 +1,105 @@
package com.patbond.patbond.creation.provider.ai;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.databind.PropertyNamingStrategies;
import com.patbond.patbond.creation.config.CreationProperties;
import com.patbond.patbond.creation.security.ServiceTokenSigner;
import feign.Request;
import feign.RequestInterceptor;
import feign.codec.Decoder;
import feign.codec.Encoder;
import feign.codec.ErrorDecoder;
import org.springframework.boot.autoconfigure.http.HttpMessageConverters;
import org.springframework.cloud.openfeign.support.ResponseEntityDecoder;
import org.springframework.cloud.openfeign.support.SpringDecoder;
import org.springframework.cloud.openfeign.support.SpringEncoder;
import org.springframework.context.annotation.Bean;
import org.springframework.http.converter.json.MappingJackson2HttpMessageConverter;
import java.util.concurrent.TimeUnit;
/**
* Feign child-context beans for the patbond-ai client — deliberately not a
* {@code @Configuration}, same reasoning as the existing auth/community Feign
* configs (a component-scanned bean would land in the parent context and be
* shadowed by the child's defaults).
*
* <p>Four things are wired here, each for a specific reason.
*/
public class AiFeignConfig {
/**
* snake_case on the wire, camelCase in Java.
*
* <p>Scoped to this client only — the contract we expose to clients stays
* camelCase, and a global naming strategy would silently reshape every other
* service's JSON.
*
* <p>Two alternatives rejected. Per-field {@code @JsonProperty}: one
* forgotten annotation silently sends a field the other end ignores, and a
* silently ignored field is the hardest failure to notice ("I passed
* duration_sec but it had no effect"). Pulling in {@code feign-jackson} for
* its own encoder: no other module needs it, and reusing Spring's converter
* stack keeps this client behaving like the rest of the codebase.
*/
private static MappingJackson2HttpMessageConverter snakeCaseConverter() {
ObjectMapper mapper = new ObjectMapper()
.setPropertyNamingStrategy(PropertyNamingStrategies.SNAKE_CASE);
return new MappingJackson2HttpMessageConverter(mapper);
}
@Bean
public Encoder aiEncoder() {
return new SpringEncoder(() -> new HttpMessageConverters(snakeCaseConverter()));
}
@Bean
public Decoder aiDecoder() {
return new ResponseEntityDecoder(
new SpringDecoder(() -> new HttpMessageConverters(snakeCaseConverter())));
}
/**
* Splits patbond-ai's two error kinds into two exceptions — see
* {@link AiErrorDecoder}. Without it, a vendor rate limit and a malformed
* request body would both surface as the same generic {@code FeignException},
* and the worker could only guess whether retrying helps.
*/
@Bean
public ErrorDecoder aiErrorDecoder() {
return new AiErrorDecoder();
}
/**
* Presents a freshly signed service-identity token on every call.
*
* <p>Signed per call, never cached: the TTL is minutes, signing is cheap, and
* a cached token outliving its purpose is how a short-lived credential
* quietly becomes a long-lived one.
*/
@Bean
public RequestInterceptor serviceTokenInterceptor(ServiceTokenSigner signer) {
return template -> template.header("Authorization", "Bearer " + signer.sign());
}
/**
* Timeouts bound the SUBMIT/QUERY round trip, <b>not the generation</b>.
*
* <p>patbond-ai's boundary is asynchronous regardless of whether the vendor
* is (ADR-023 §4): submit only enqueues. Inflating these to "wait for the
* video" would reintroduce held connections and the ambiguous-billing window
* the async boundary exists to remove.
*
* <p>{@code followRedirects = false}: an internal service has no business
* redirecting us, and following one would send the service token to whatever
* host the redirect names.
*/
@Bean
public Request.Options aiRequestOptions(CreationProperties properties) {
CreationProperties.AiService config = properties.getAiService();
return new Request.Options(
config.getConnectTimeout().toMillis(), TimeUnit.MILLISECONDS,
config.getReadTimeout().toMillis(), TimeUnit.MILLISECONDS,
false);
}
}
@@ -0,0 +1,35 @@
package com.patbond.patbond.creation.provider.ai;
import org.springframework.cloud.openfeign.FeignClient;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestBody;
/**
* Feign client for patbond-ai's internal generation API.
*
* <p>Static direct URL per ADR-002 (no service discovery in the MVP), same as
* the existing author-profile client in patbond-community.
*
* <p>The path parameter is the <b>vendor's</b> task id, not our jobId. That is
* patbond-ai's choice and a deliberate one: it holds no state, so the vendor id
* is the only thing it can ask the vendor about. Using our jobId would force it
* to keep a mapping, and would create a window where we hold an id that cannot
* query anything.
*/
@FeignClient(
name = "patbond-ai",
// 占位符带默认值:Feign 在**上下文启动时**解析 url,无默认值时任何没有
// 显式配置该属性的上下文都会启动失败——包括与 AI 无关的仓储测试。
// 与 CreationProperties.AiService 的默认值保持一致。
url = "${patbond.creation.ai-service.url:http://127.0.0.1:8000}",
configuration = AiFeignConfig.class)
public interface AiGenerationClient {
@PostMapping("/internal/generation/tasks")
AiContract.SubmitResponse submit(@RequestBody AiContract.SubmitRequest request);
@GetMapping("/internal/generation/tasks/{providerRequestId}")
AiContract.TaskResponse get(@PathVariable("providerRequestId") String providerRequestId);
}
@@ -0,0 +1,26 @@
package com.patbond.patbond.creation.provider.ai;
import com.patbond.patbond.creation.provider.GenerationProvider.PollResult;
/**
* The vendor said no. Carries the classification so the worker can decide
* without interpreting vendor codes.
*
* <p>Distinct from {@link AiServiceException} because the reactions differ: this
* one may be worth retrying (see {@link PollResult.Failure#retry()}), that one
* means we sent something wrong.
*/
public class AiProviderException extends RuntimeException {
private final PollResult.Failure failure;
AiProviderException(AiErrorDecoder.PollResultFailure raw) {
super(raw.message() == null ? raw.providerCode() : raw.message());
this.failure = new PollResult.Failure(
raw.retry(), raw.userFacing(), raw.providerCode(), raw.message());
}
public PollResult.Failure failure() {
return failure;
}
}
@@ -0,0 +1,45 @@
package com.patbond.patbond.creation.provider.ai;
/**
* Something went wrong between us and patbond-ai: our request body did not match
* the contract, the service token was rejected, the task id is unknown, or the
* adapter is not assembled there.
*
* <p><b>Mostly not worth retrying</b> — retrying a malformed request never helps.
* The two exceptions are {@code internal_error} and {@code provider_unavailable}:
* the first may be transient, the second usually means credentials are not
* configured yet and will start working once they are.
*/
public class AiServiceException extends RuntimeException {
/** Contract code: unauthenticated / invalid_request / task_not_found / … */
private final String code;
private final int httpStatus;
AiServiceException(int httpStatus, String code, String message) {
super(message == null ? code : message);
this.httpStatus = httpStatus;
this.code = code;
}
public String code() {
return code;
}
public int httpStatus() {
return httpStatus;
}
/** The vendor no longer has this task — stop polling rather than back off. */
public boolean isTaskNotFound() {
return "task_not_found".equals(code);
}
/**
* @return true when the condition may clear on its own (a transient bug on
* their side, or credentials that are about to be configured)
*/
public boolean mayClearOnItsOwn() {
return "internal_error".equals(code) || "provider_unavailable".equals(code);
}
}
@@ -49,7 +49,8 @@ public class GenerationJobRepository {
"input_asset_id", "output_asset_id", "prompt", "negative_prompt",
"width_px", "height_px", "duration_ms", "status", "progress",
"attempt_count", "max_attempts", "provider_code_snapshot",
"provider_request_id", "error_code", "error_message",
"provider_model_snapshot", "provider_request_id",
"error_code", "error_message",
"created_at", "first_attempt_at", "started_at", "completed_at",
"lease_expires_at");
@@ -419,6 +420,7 @@ public class GenerationJobRepository {
rs.getShort("attempt_count"),
rs.getShort("max_attempts"),
rs.getString("provider_code_snapshot"),
rs.getString("provider_model_snapshot"),
rs.getString("provider_request_id"),
rs.getString("error_code"),
rs.getString("error_message"),
@@ -19,6 +19,22 @@ public final class RequestHashes {
private RequestHashes() {
}
/**
* SHA-256 hex of a value, for identifiers that must be stable but must not
* carry the original. Used for the vendor's {@code safety_identifier}: it
* needs a per-user constant, and a hash serves that without handing over our
* user ids (ADR-023 §5).
*/
public static String hex(String value) {
byte[] digest = sha256(value);
StringBuilder out = new StringBuilder(digest.length * 2);
for (byte b : digest) {
out.append(Character.forDigit((b >> 4) & 0xF, 16));
out.append(Character.forDigit(b & 0xF, 16));
}
return out.toString();
}
public static byte[] sha256(String canonical) {
try {
return MessageDigest.getInstance("SHA-256")
@@ -4,6 +4,8 @@ import com.patbond.patbond.creation.config.CreationProperties;
import com.patbond.patbond.creation.config.ProviderRegistry;
import com.patbond.patbond.creation.domain.GenerationJob;
import com.patbond.patbond.creation.provider.GenerationProvider;
import com.patbond.patbond.creation.provider.ai.AiProviderException;
import com.patbond.patbond.creation.provider.ai.AiServiceException;
import com.patbond.patbond.creation.repository.GenerationJobRepository;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -80,9 +82,28 @@ public class GenerationWorker {
for (GenerationJob job : claimed) {
try {
submitOne(job);
} catch (AiProviderException e) {
// 服务商说不行。分类已由适配层给出,直接据此决定。
log.warn("提交任务 {} 被服务商拒绝 code={} retry={}", job.id(),
e.failure().providerCode(), e.failure().retry());
failOrRetry(job, e.failure());
} catch (AiServiceException e) {
// 我方与 patbond-ai 之间的问题:请求体不合契约、服务票被拒、适配器
// 未装配。**大多不该重试**——重试一个拼错的请求永远不会变好。
log.error("提交任务 {} 失败于内部接口 code={}", job.id(), e.code(), e);
failOrRetry(job, new GenerationProvider.PollResult.Failure(
e.mayClearOnItsOwn()
? GenerationProvider.Retry.BACKOFF
: GenerationProvider.Retry.NEVER,
GenerationProvider.UserFacing.SERVER_ERROR,
"AI_" + (e.code() == null ? "UNKNOWN" : e.code().toUpperCase()),
e.getMessage()));
} catch (RuntimeException e) {
log.error("Submitting generation job {} failed", job.id(), e);
failOrRetry(job, "PROVIDER_SUBMIT_FAILED", e.getMessage(), true);
log.error("提交任务 {} 时发生未预期异常", job.id(), e);
failOrRetry(job, new GenerationProvider.PollResult.Failure(
GenerationProvider.Retry.BACKOFF,
GenerationProvider.UserFacing.SERVER_ERROR,
"SUBMIT_UNEXPECTED", e.getMessage()));
}
}
}
@@ -108,9 +129,11 @@ public class GenerationWorker {
return;
}
String uploadTarget = uploadTargets.presignedPutFor(job);
GenerationProvider.SubmitResult result =
provider.submit(job, uploadTarget, job.id().toString());
// uploadTarget 传 null:产物转存阻塞于对象存储决策(ADR-023 §7),而内部
// 契约把该字段标为可选未启用。**这里刻意不调 uploadTargets**——那个方法
// 目前会抛异常,在 submit 之前调它会让整条链一步都跑不动。启用转存时在此
// 取回预签名地址即可,接口签名无需改动。
GenerationProvider.SubmitResult result = provider.submit(job, null);
int recorded = repository.recordProviderRequest(job.id(), result.providerRequestId());
if (recorded == 0) {
// Another attempt recorded an id first; ours is a duplicate
@@ -158,18 +181,23 @@ public class GenerationWorker {
}
}
case SUCCEEDED -> {
// The artifact is already in object storage (the provider PUT it
// to the presigned target). Turning the object key into a
// media.assets row is patbond-user's job — ADR-017 keeps the
// media write flow there — so this step waits on the internal
// media API, which in turn waits on the storage decision.
// 产物已生成,元数据(尺寸、时长、封面、计费 token)都已在手——但把它
// 变成 media.assets 行是 patbond-user 的职责(ADR-017 把媒体写入流程
// 留在那里),需先落地其 /internal/media/**,而该端点的返回形态取决于
// **挂起中的**对象存储决策(ADR-023 §7)。
//
// 这里有时限:服务商产物 URL 仅 24 小时有效,过期即死链。因此
// 「先跑通链路、转存以后再说」这个状态下跑出来的产物留不住。
log.error("产物入库未实现,任务 {} 的产物将在 {} 秒后失效:"
+ "assetUrl={} cover={} {}x{} durationMs={} tokens={}",
job.id(), result.assetUrlExpiresInSec(), result.assetUrl(),
result.coverUrl(), result.widthPx(), result.heightPx(),
result.durationMs(), result.billingTokens());
throw new UnsupportedOperationException(
"产物入库未实现:objectKey=" + result.objectKey()
+ "。需先落地 patbond-user 的 /internal/media/** 确认端点,"
+ "且其返回形态取决于对象存储方案(ADR-023 §7 挂起)。");
"产物入库未实现(ADR-023 §7 对象存储决策挂起):job=" + job.id());
}
case FAILED -> failOrRetry(job, result.errorCode(), result.errorMessage(),
result.retryable());
case FAILED -> failOrRetry(job, result.failure());
case CANCELLED -> log.info("任务 {} 已在服务商侧取消", job.id());
}
}
@@ -192,20 +220,35 @@ public class GenerationWorker {
// ------------------------------------------------------------- helpers
/**
* Terminal failure or another attempt, decided by the retry budget and
* whether the failure is retryable at all. A rejected prompt is not worth
* retrying; a vendor 5xx is.
* Terminal failure or another attempt.
*
* <p>The decision uses the adapter's three-valued {@code retry}, not a
* boolean. The distinction that matters: a rate limit clears in seconds
* ({@code BACKOFF}), an exhausted quota clears tomorrow ({@code QUOTA}).
* Treating the latter as retryable burns the whole retry budget for nothing
* and still fails — and meanwhile the user watches a spinner instead of the
* "quota exhausted" sheet that would tell them what to do (ADR-023 §9).
*/
private void failOrRetry(GenerationJob job, String errorCode, String errorMessage,
boolean retryable) {
if (!retryable || job.isLastAttempt()) {
repository.markFailed(job.id(), errorCode, truncate(errorMessage));
private void failOrRetry(GenerationJob job, GenerationProvider.PollResult.Failure failure) {
if (!failure.retry().worthRetrying() || job.isLastAttempt()) {
repository.markFailed(job.id(), errorCodeOf(failure), truncate(failure.message()));
return;
}
Duration backoff = properties.getWorker().retryBackoffFor(job.attemptCount());
repository.requeueWithBackoff(job.id(), backoff);
}
/**
* The code persisted on the job row, and thus what the client eventually
* sees. Uses the adapter's {@code userFacing} category rather than the raw
* vendor code: the vendor's vocabulary must not leak into our data — it would
* change when the provider changes, and clients would end up branching on it.
* The raw code stays in the error message for troubleshooting.
*/
private static String errorCodeOf(GenerationProvider.PollResult.Failure failure) {
return failure.userFacing().name();
}
/** error_message is varchar(1000); a vendor stack trace must not blow the insert. */
private static String truncate(String message) {
if (message == null) {
@@ -0,0 +1,399 @@
package com.patbond.patbond.creation.provider.ai;
import com.patbond.patbond.creation.TestcontainersConfiguration;
import com.patbond.patbond.creation.domain.GenerationJob;
import com.patbond.patbond.creation.domain.JobStatus;
import com.patbond.patbond.creation.domain.MediaKind;
import com.patbond.patbond.creation.provider.AiServiceGenerationProvider;
import com.patbond.patbond.creation.provider.GenerationProvider;
import com.patbond.patbond.creation.provider.GenerationProvider.PollResult;
import com.patbond.patbond.creation.provider.GenerationProvider.Retry;
import com.patbond.patbond.creation.provider.GenerationProvider.UserFacing;
import com.sun.net.httpserver.HttpServer;
import org.junit.jupiter.api.AfterAll;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.context.annotation.Import;
import org.springframework.test.context.DynamicPropertyRegistry;
import org.springframework.test.context.DynamicPropertySource;
import org.springframework.test.context.TestPropertySource;
import java.io.IOException;
import java.io.OutputStream;
import java.net.InetSocketAddress;
import java.nio.charset.StandardCharsets;
import java.util.UUID;
import java.util.concurrent.ThreadLocalRandom;
import java.util.concurrent.atomic.AtomicReference;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
/**
* The real Feign wiring against an in-test HTTP server standing in for
* patbond-ai: snake_case serialisation both ways, the service-token interceptor,
* and — the important part — <b>the two error kinds landing on two different
* exceptions</b>.
*
* <p>Why the error split deserves wire-level tests rather than unit tests on the
* decoder: if the split silently collapses (one generic FeignException), the
* worker can only guess whether retrying helps. A quota-exhausted job would then
* burn its whole retry budget and still fail, while the user watches a spinner
* instead of being told to top up.
*/
@SpringBootTest
@Import(TestcontainersConfiguration.class)
@TestPropertySource(properties = "patbond.creation.worker.enabled=false")
class AiGenerationClientWireTest {
private static HttpServer server;
private static final AtomicReference<String> responseBody = new AtomicReference<>("{}");
private static final AtomicReference<Integer> responseStatus = new AtomicReference<>(200);
private static final AtomicReference<String> lastRequestBody = new AtomicReference<>("");
private static final AtomicReference<String> lastAuthorization = new AtomicReference<>("");
private static final AtomicReference<String> lastPath = new AtomicReference<>("");
@Autowired
private AiGenerationClient client;
@DynamicPropertySource
static void aiService(DynamicPropertyRegistry registry) throws IOException {
int port = ThreadLocalRandom.current().nextInt(20000, 45000);
server = HttpServer.create(new InetSocketAddress("127.0.0.1", port), 0);
server.createContext("/internal/generation", exchange -> {
lastPath.set(exchange.getRequestURI().getPath());
lastAuthorization.set(
String.valueOf(exchange.getRequestHeaders().getFirst("Authorization")));
lastRequestBody.set(
new String(exchange.getRequestBody().readAllBytes(), StandardCharsets.UTF_8));
byte[] body = responseBody.get().getBytes(StandardCharsets.UTF_8);
exchange.getResponseHeaders().add("Content-Type", "application/json");
exchange.sendResponseHeaders(responseStatus.get(), body.length);
try (OutputStream out = exchange.getResponseBody()) {
out.write(body);
}
});
server.start();
registry.add("patbond.creation.ai-service.url", () -> "http://127.0.0.1:" + port);
// 服务票需要一把私钥才能签出来;测试期现场生成。
registry.add("patbond.creation.service-token.private-key", TestKeys::privateKeyPem);
}
@AfterAll
static void stopServer() {
if (server != null) {
server.stop(0);
}
}
@BeforeEach
void reset() {
responseStatus.set(200);
responseBody.set("{}");
lastRequestBody.set("");
}
// ------------------------------------------------------- 序列化与鉴权
@Test
void requestBodyIsSerialisedInSnakeCase() {
responseBody.set("{\"provider_request_id\":\"cgt-1\"}");
client.submit(new AiContract.SubmitRequest(
"job-1", "video", "doubao-seedance-2-5-260628", "一只猫",
null, null, "720p", 5, true, false, "hash", null, null));
String sent = lastRequestBody.get();
assertThat(sent).contains("\"job_id\"", "\"media_kind\"", "\"duration_sec\"",
"\"generate_audio\"", "\"safety_identifier\"");
// camelCase 一个都不该出现——出现即意味着命名策略没生效,而症状是对端
// 静默忽略那些字段(extra=forbid 下则是 400),两者都难查。
assertThat(sent).doesNotContain("jobId", "mediaKind", "durationSec");
}
@Test
void unsetOptionalFieldsAreOmittedNotSentAsNull() {
// patbond-ai 侧 extra="forbid" 且各字段有文档化的默认值;显式 null 会
// 覆盖掉那些默认值。
responseBody.set("{\"provider_request_id\":\"cgt-1\"}");
client.submit(new AiContract.SubmitRequest(
"job-1", "video", "m", "prompt",
null, null, null, null, null, null, null, null, null));
assertThat(lastRequestBody.get()).doesNotContain("null");
}
@Test
void responseIsDeserialisedFromSnakeCase() {
responseBody.set("""
{"provider_request_id":"cgt-1","state":"succeeded","progress":100,
"asset_url":"https://v/out.mp4","asset_url_expires_in_sec":86400,
"cover_url":"https://v/f.png","ratio":"16:9","resolution":"720p",
"width_px":1280,"height_px":720,"duration_ms":5542,
"output_format":"mp4","billing_tokens":108900,"failure":null}
""");
AiContract.TaskResponse response = client.get("cgt-1");
assertThat(response.assetUrl()).isEqualTo("https://v/out.mp4");
assertThat(response.assetUrlExpiresInSec()).isEqualTo(86400);
assertThat(response.widthPx()).isEqualTo(1280);
assertThat(response.durationMs()).isEqualTo(5542);
assertThat(response.billingTokens()).isEqualTo(108900);
}
@Test
void everyCallCarriesAFreshlySignedServiceToken() {
responseBody.set("{\"provider_request_id\":\"cgt-1\"}");
client.submit(submitRequest());
String first = lastAuthorization.get();
client.submit(submitRequest());
String second = lastAuthorization.get();
assertThat(first).startsWith("Bearer ");
// 每次现签、不缓存:短期凭证一旦被缓存复用,就悄悄变成了长期凭证。
// jti 与 iat 不同,故两张票的字节不同。
assertThat(second).isNotEqualTo(first);
}
@Test
void pathParameterIsTheVendorTaskId() {
responseBody.set("{\"provider_request_id\":\"cgt-abc\",\"state\":\"running\",\"progress\":0}");
client.get("cgt-abc");
assertThat(lastPath.get()).isEqualTo("/internal/generation/tasks/cgt-abc");
}
// --------------------------------------------- 两类错误落在两个异常上
@Test
void providerErrorBecomesAiProviderExceptionCarryingTheClassification() {
responseStatus.set(502);
responseBody.set("""
{"error":{"kind":"provider","retry":"backoff",
"user_facing":"rate_limited",
"provider_code":"RateLimitExceeded.EndpointRPMExceeded",
"message":"RPM limit exceeded"}}
""");
assertThatThrownBy(() -> client.submit(submitRequest()))
.isInstanceOf(AiProviderException.class)
.satisfies(thrown -> {
PollResult.Failure failure = ((AiProviderException) thrown).failure();
assertThat(failure.retry()).isEqualTo(Retry.BACKOFF);
assertThat(failure.userFacing()).isEqualTo(UserFacing.RATE_LIMITED);
assertThat(failure.providerCode())
.isEqualTo("RateLimitExceeded.EndpointRPMExceeded");
});
}
@Test
void quotaExhaustedIsNotWorthRetryingUnlikeARateLimit() {
// 同为服务商 429,但一个等几秒、一个等到明天。若两者都按可重试处理,
// 配额耗尽的任务会白烧完整个重试预算再失败。
responseStatus.set(502);
responseBody.set("""
{"error":{"kind":"provider","retry":"quota",
"user_facing":"quota_exhausted","provider_code":"QuotaExceeded",
"message":"exhausted its free trial quota"}}
""");
assertThatThrownBy(() -> client.submit(submitRequest()))
.isInstanceOf(AiProviderException.class)
.satisfies(thrown -> {
PollResult.Failure failure = ((AiProviderException) thrown).failure();
assertThat(failure.retry()).isEqualTo(Retry.QUOTA);
assertThat(failure.retry().worthRetrying()).isFalse();
assertThat(failure.userFacing()).isEqualTo(UserFacing.QUOTA_EXHAUSTED);
});
}
@Test
void serviceErrorBecomesAiServiceExceptionWithItsContractCode() {
responseStatus.set(400);
responseBody.set("""
{"error":{"kind":"service","code":"invalid_request",
"message":"duration_sec: 模型 x 的时长需在 [4, 15] 秒内"}}
""");
assertThatThrownBy(() -> client.submit(submitRequest()))
.isInstanceOf(AiServiceException.class)
.satisfies(thrown -> {
AiServiceException error = (AiServiceException) thrown;
assertThat(error.code()).isEqualTo("invalid_request");
assertThat(error.mayClearOnItsOwn()).isFalse();
assertThat(error.getMessage()).contains("[4, 15]");
});
}
@Test
void taskNotFoundIsDistinguishableSoThePollerCanStop() {
responseStatus.set(404);
responseBody.set("""
{"error":{"kind":"service","code":"task_not_found",
"message":"服务商无此任务"}}
""");
assertThatThrownBy(() -> client.get("cgt-gone"))
.isInstanceOf(AiServiceException.class)
.satisfies(thrown ->
assertThat(((AiServiceException) thrown).isTaskNotFound()).isTrue());
}
@Test
void providerUnavailableMayClearOnItsOwn() {
// 通常是对端还没配服务商凭证——配好就能用,故值得重试。
responseStatus.set(503);
responseBody.set("""
{"error":{"kind":"service","code":"provider_unavailable",
"message":"没有可用的 video 生成适配器"}}
""");
assertThatThrownBy(() -> client.submit(submitRequest()))
.isInstanceOf(AiServiceException.class)
.satisfies(thrown ->
assertThat(((AiServiceException) thrown).mayClearOnItsOwn()).isTrue());
}
@Test
void anUnrecognisedErrorBodyIsTreatedAsRetryableAndSaysSo() {
// 打到错误地址时常见:网关返回 HTML。静默归类会让排查走错方向。
responseStatus.set(502);
responseBody.set("<html>bad gateway</html>");
assertThatThrownBy(() -> client.get("cgt-1"))
.isInstanceOf(AiProviderException.class)
.satisfies(thrown -> {
PollResult.Failure failure = ((AiProviderException) thrown).failure();
assertThat(failure.retry()).isEqualTo(Retry.BACKOFF);
assertThat(failure.providerCode()).isEqualTo("AI_RESPONSE_UNRECOGNISED");
});
}
@Test
void unknownRetryValueLeansTowardsRetrying() {
// 契约可能先长出一个取值,代码后知道。把可恢复的抖动判成用户可见的失败
// 是更糟的错误,而重试预算本身有上限。
responseStatus.set(502);
responseBody.set("""
{"error":{"kind":"provider","retry":"brand_new_strategy",
"user_facing":"upstream_unavailable","provider_code":"X","message":"m"}}
""");
assertThatThrownBy(() -> client.submit(submitRequest()))
.isInstanceOf(AiProviderException.class)
.satisfies(thrown -> assertThat(
((AiProviderException) thrown).failure().retry()).isEqualTo(Retry.BACKOFF));
}
// ------------------------------------------------- 适配器层的端到端映射
@Test
void adapterMapsSucceededTaskIntoAFullPollResult() {
responseBody.set("""
{"provider_request_id":"cgt-1","state":"succeeded","progress":100,
"asset_url":"https://v/out.mp4","asset_url_expires_in_sec":86400,
"cover_url":"https://v/f.png","ratio":"16:9","resolution":"720p",
"width_px":1280,"height_px":720,"duration_ms":5542,
"output_format":"mp4","billing_tokens":108900}
""");
PollResult result = new AiServiceGenerationProvider(client).poll(job(), "cgt-1");
assertThat(result.state()).isEqualTo(PollResult.State.SUCCEEDED);
assertThat(result.assetUrl()).isEqualTo("https://v/out.mp4");
// 时效随 URL 一起带上,故上层没有理由把服务商地址当永久存储。
assertThat(result.assetUrlExpiresInSec()).isEqualTo(86400);
assertThat(result.widthPx()).isEqualTo(1280);
assertThat(result.billingTokens()).isEqualTo(108900);
assertThat(result.failure()).isNull();
}
@Test
void adapterSendsTheRowsFrozenModelSnapshotNotTheLiveCatalog() {
// 任务记录里写的模型必须与实际调用的一致,否则事后无法对账。
responseBody.set("{\"provider_request_id\":\"cgt-1\"}");
new AiServiceGenerationProvider(client).submit(job(), null);
assertThat(lastRequestBody.get()).contains("doubao-seedance-2-5-260628");
}
@Test
void adapterHashesTheUserIdForTheVendorInsteadOfSendingIt() {
responseBody.set("{\"provider_request_id\":\"cgt-1\"}");
GenerationJob job = job();
new AiServiceGenerationProvider(client).submit(job, null);
String sent = lastRequestBody.get();
assertThat(sent).contains("safety_identifier");
// 明文 userId 绝不外发(ADR-023 §5 数据最小化)。
assertThat(sent).doesNotContain(job.userId().toString());
}
@Test
void adapterTurnsVendorTaskGoneIntoATerminalFailure() {
responseStatus.set(404);
responseBody.set("""
{"error":{"kind":"service","code":"task_not_found","message":"gone"}}
""");
PollResult result = new AiServiceGenerationProvider(client).poll(job(), "cgt-gone");
// 服务商已无此任务 → 没有什么可轮询了,终态失败而不是永远退避。
assertThat(result.state()).isEqualTo(PollResult.State.FAILED);
assertThat(result.failure().retry()).isEqualTo(Retry.NEVER);
assertThat(result.failure().providerCode()).isEqualTo("PROVIDER_TASK_GONE");
}
@Test
void adapterProviderCodeMatchesTheSeededCatalogValue() {
// 三处必须一致:本常量、patbond-ai 的 PROVIDER_CODE、V7 种子的 provider_code。
// 不一致时任务永远停在排队,而那看起来像队列坏了。
assertThat(new AiServiceGenerationProvider(client).providerCode())
.isEqualTo("volcengine-seedance");
}
// --------------------------------------------------------------- helpers
private static AiContract.SubmitRequest submitRequest() {
return new AiContract.SubmitRequest("job-1", "video", "m", "prompt",
null, null, null, null, null, null, null, null, null);
}
private static GenerationJob job() {
return new GenerationJob(
UUID.randomUUID(), UUID.randomUUID(), null, MediaKind.VIDEO,
UUID.randomUUID(), null, null, null, "一只猫", null,
null, null, 5000L, JobStatus.RUNNING, (short) 0, (short) 1, (short) 3,
"volcengine-seedance", "doubao-seedance-2-5-260628", null,
null, null, null, null, null, null, null);
}
/** 测试期临时 RSA 私钥,供服务票签发。 */
private static final class TestKeys {
private static final String PEM;
static {
try {
var generator = java.security.KeyPairGenerator.getInstance("RSA");
generator.initialize(2048);
var keyPair = generator.generateKeyPair();
PEM = "-----BEGIN PRIVATE KEY-----\n"
+ java.util.Base64.getMimeEncoder(64, new byte[]{'\n'})
.encodeToString(keyPair.getPrivate().getEncoded())
+ "\n-----END PRIVATE KEY-----\n";
} catch (Exception e) {
throw new IllegalStateException(e);
}
}
static String privateKeyPem() {
return PEM;
}
}
}