diff --git a/backend/internal/provider/adobe/client.go b/backend/internal/provider/adobe/client.go index 769088f..8bc74d7 100644 --- a/backend/internal/provider/adobe/client.go +++ b/backend/internal/provider/adobe/client.go @@ -37,8 +37,21 @@ var ( ErrQuotaExhausted = errors.New("adobe quota exhausted") ErrTemporaryUpstream = errors.New("adobe upstream temporary error") ErrDeadUpstream = errors.New("adobe upstream fatal error") + // ErrContentRejected is Adobe's content-safety filter refusing the prompt or + // the generated image (HTTP 451 image_unsafe). It is the prompt's fault, not + // the account's — every account rejects the same content — so the caller must + // surface it as-is without penalizing/killing the account or failing over. + ErrContentRejected = errors.New("adobe content rejected") ) +// isContentRejection reports whether an Adobe response (status + body) is a +// content-safety refusal rather than a genuine upstream/account failure. Adobe +// returns HTTP 451 with an "*_unsafe" error_code when moderation blocks the +// prompt or the produced image. +func isContentRejection(status int, body string) bool { + return status == 451 && strings.Contains(body, "unsafe") +} + var profileURLs = []string{ "https://ims-na1.adobelogin.com/ims/profile/v1", "https://adobeid-na1.services.adobe.com/ims/profile/v1", @@ -189,10 +202,15 @@ func (c *Client) GenerateImage(ctx context.Context, token, modelID, prompt, aspe } lastBody = respBody lastErr = err - if errors.Is(err, ErrAuth) || errors.Is(err, ErrQuotaExhausted) { + if errors.Is(err, ErrAuth) || errors.Is(err, ErrQuotaExhausted) || errors.Is(err, ErrContentRejected) { return nil, nil, err } } + // Content-safety refusal: the prompt/image is blocked, retrying other payloads + // or accounts is pointless — surface it as-is so the pool doesn't fail over. + if errors.Is(lastErr, ErrContentRejected) { + return nil, nil, fmt.Errorf("%w: adobe submit: %s", ErrContentRejected, clip(lastBody, 300)) + } // Preserve the temporary classification so the pool retries (overload / 5xx / // rate-limit) instead of failing the request outright. if errors.Is(lastErr, ErrDeadUpstream) { @@ -471,6 +489,9 @@ func (c *Client) submitImage(ctx context.Context, sess *tlsSession, token, promp if b := string(respBody); strings.Contains(b, "system under load") || strings.Contains(b, "timeout_error") { return respBody, "", ErrTemporaryUpstream } + if isContentRejection(resp.StatusCode, string(respBody)) { + return respBody, "", ErrContentRejected + } if resp.StatusCode == 429 || resp.StatusCode == 451 || resp.StatusCode >= 500 { return respBody, "", ErrDeadUpstream } @@ -537,6 +558,9 @@ func (c *Client) pollImage(ctx context.Context, sess *tlsSession, token, pollURL if b := string(body); strings.Contains(b, "system under load") || strings.Contains(b, "timeout_error") { return nil, nil, ErrTemporaryUpstream } + if isContentRejection(resp.StatusCode, string(body)) { + return nil, nil, fmt.Errorf("%w: %s", ErrContentRejected, clip(body, 300)) + } if resp.StatusCode == 429 || resp.StatusCode == 451 || resp.StatusCode >= 500 { return nil, nil, ErrDeadUpstream } @@ -640,6 +664,9 @@ func (c *Client) submitVideo(ctx context.Context, sess *tlsSession, token, endpo // it's a bad token, a missing scope, or a WAF/fingerprint block. return respBody, "", fmt.Errorf("%w (%d %s: %s)", ErrAuth, resp.StatusCode, resp.Header.Get("x-access-error"), clip(respBody, 300)) } + if isContentRejection(resp.StatusCode, string(respBody)) { + return respBody, "", ErrContentRejected + } if resp.StatusCode == 408 || resp.StatusCode == 429 || resp.StatusCode == 451 || resp.StatusCode >= 500 { return respBody, "", ErrDeadUpstream } @@ -714,6 +741,9 @@ func (c *Client) pollVideo(ctx context.Context, sess *tlsSession, token, pollURL if b := string(body); strings.Contains(b, "system under load") || strings.Contains(b, "timeout_error") { return nil, nil, ErrTemporaryUpstream } + if isContentRejection(resp.StatusCode, string(body)) { + return nil, nil, fmt.Errorf("%w: %s", ErrContentRejected, clip(body, 300)) + } if resp.StatusCode == 429 || resp.StatusCode == 451 || resp.StatusCode >= 500 { return nil, nil, ErrDeadUpstream } diff --git a/backend/internal/service/v1.go b/backend/internal/service/v1.go index 67508c5..50a5523 100644 --- a/backend/internal/service/v1.go +++ b/backend/internal/service/v1.go @@ -1461,6 +1461,7 @@ func (s *V1Service) tryAccount(ctx context.Context, eventID, pool string, token _ = s.tokens.TouchLastUsed(ctx, token.ID) authRefreshed := false tempAttempts := 0 + fatalAttempts := 0 for { data, err := attempt(token) if err == nil { @@ -1488,20 +1489,33 @@ func (s *V1Service) tryAccount(ctx context.Context, eventID, pool string, token s.markTokenFailure(ctx, pool, token, kind, true, false) return nil, err, true, false } - if isDead { + // Fatal / (temporary under adobe's failover policy) upstream error. + if isDead || (isTemp && tempFailover) { + if tempFailover { + // Ops policy (adobe): NEVER kill on these upstream errors — a + // genuinely bad account and a transient Adobe blip (429/5xx/ + // overload) look the same, and killing wipes healthy accounts. + // First retry the SAME account a few times; if still failing, just + // record the failure and fail over to the next account (no + // disable/dead). The 4th return value caps how many accounts one + // request may burn this way (maxTempDeadAccounts) so a pool-wide + // blip can't fan a single request across the whole pool. + fatalAttempts++ + if fatalAttempts < maxSameAccountAttempts { + select { + case <-time.After(time.Duration(fatalAttempts) * time.Second): + continue + case <-ctx.Done(): + return nil, err, false, false + } + } + s.markTokenFailure(ctx, pool, token, kind, false, false) + return nil, err, true, true + } s.markTokenDead(ctx, pool, token, kind) return nil, err, true, true } if isTemp { - if tempFailover { - // Ops policy (adobe): treat a temporary upstream error the same as - // a fatal one — disable+mark the account dead and fail over to the - // next. The pool driver caps how many accounts one request may burn - // this way (maxTempDeadAccounts) so a pool-wide blip can't fan a - // single request across the whole pool in one shot. - s.markTokenDead(ctx, pool, token, kind) - return nil, err, true, true - } tempAttempts++ if tempAttempts < maxSameAccountAttempts { // Short linear backoff (1s, 2s) so an overloaded/rate-limited upstream