From f9e72c0168daf12b495f0ac31dd37e26f2c9bd3b Mon Sep 17 00:00:00 2001 From: chiyi Date: Fri, 3 Jul 2026 18:12:46 +0800 Subject: [PATCH] =?UTF-8?q?=E4=BF=AE=E5=A4=8Dbug?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- backend/internal/provider/adobe/client.go | 32 +++++++---- backend/internal/provider/grok/client.go | 32 +++++------ backend/internal/service/maintenance.go | 18 ++++-- backend/internal/service/tokens.go | 69 +++++++++++++++++++++-- backend/internal/service/v1.go | 42 +++++++++----- 5 files changed, 142 insertions(+), 51 deletions(-) diff --git a/backend/internal/provider/adobe/client.go b/backend/internal/provider/adobe/client.go index 3268095..c56673c 100644 --- a/backend/internal/provider/adobe/client.go +++ b/backend/internal/provider/adobe/client.go @@ -19,22 +19,23 @@ import ( ) const ( - submitURL = "https://firefly-3p.ff.adobe.io/v2/3p-images/generate-async" + submitURL = "https://firefly-3p.ff.adobe.io/v2/3p-images/generate-async" image5SubmitURL = "https://image-v5.ff.adobe.io/v1/images/generate-async" - videoSubmitURL = "https://firefly-3p.ff.adobe.io/v2/3p-videos/generate-async" + videoSubmitURL = "https://firefly-3p.ff.adobe.io/v2/3p-videos/generate-async" // Firefly-native video model (project id "firefly-video"): distinct host, // submit path and storage host from the 3p (veo/luma) video flow. fireflyVideoSubmitURL = "https://video-v1.ff.adobe.io/v2/videos/generate" fireflyVideoUploadURL = "https://video-v1.ff.adobe.io/v2/storage/image" - uploadURL = "https://firefly-3p.ff.adobe.io/v2/storage/image" - creditsURL = "https://firefly.adobe.io/v1/credits/balance" - creditsAPIKey = "SunbreakWebUI1" + uploadURL = "https://firefly-3p.ff.adobe.io/v2/storage/image" + creditsURL = "https://firefly.adobe.io/v1/credits/balance" + creditsAPIKey = "SunbreakWebUI1" ) var ( ErrAuth = errors.New("adobe auth failed") ErrQuotaExhausted = errors.New("adobe quota exhausted") ErrTemporaryUpstream = errors.New("adobe upstream temporary error") + ErrDeadUpstream = errors.New("adobe upstream fatal error") ) var profileURLs = []string{ @@ -193,6 +194,9 @@ func (c *Client) GenerateImage(ctx context.Context, token, modelID, prompt, aspe } // Preserve the temporary classification so the pool retries (overload / 5xx / // rate-limit) instead of failing the request outright. + if errors.Is(lastErr, ErrDeadUpstream) { + return nil, nil, fmt.Errorf("%w: adobe submit: %s", ErrDeadUpstream, clip(lastBody, 300)) + } if errors.Is(lastErr, ErrTemporaryUpstream) { return nil, nil, fmt.Errorf("%w: adobe submit: %s", ErrTemporaryUpstream, clip(lastBody, 300)) } @@ -461,14 +465,14 @@ func (c *Client) submitImage(ctx context.Context, client tlsclient.HttpClient, t } return respBody, "", fmt.Errorf("%w (submit %d %s: %s)", ErrAuth, resp.StatusCode, resp.Header.Get("x-access-error"), clip(respBody, 300)) } - if resp.StatusCode == 429 || resp.StatusCode == 451 || resp.StatusCode >= 500 { - return respBody, "", ErrTemporaryUpstream - } // "system under load" / timeout_error = adobe rate-limit/overload (can come on a // non-5xx) — treat as temporary so the pool retries instead of failing. if b := string(respBody); strings.Contains(b, "system under load") || strings.Contains(b, "timeout_error") { return respBody, "", ErrTemporaryUpstream } + if resp.StatusCode == 429 || resp.StatusCode == 451 || resp.StatusCode >= 500 { + return respBody, "", ErrDeadUpstream + } if resp.StatusCode != 200 { return respBody, "", errors.New("submit rejected") } @@ -529,9 +533,12 @@ func (c *Client) pollImage(ctx context.Context, client tlsclient.HttpClient, tok if readErr != nil { return nil, nil, readErr } - if resp.StatusCode == 429 || resp.StatusCode == 451 || resp.StatusCode >= 500 { + if b := string(body); strings.Contains(b, "system under load") || strings.Contains(b, "timeout_error") { return nil, nil, ErrTemporaryUpstream } + if resp.StatusCode == 429 || resp.StatusCode == 451 || resp.StatusCode >= 500 { + return nil, nil, ErrDeadUpstream + } if resp.StatusCode != 200 { return nil, nil, fmt.Errorf("adobe poll failed: %d %s", resp.StatusCode, clip(body, 300)) } @@ -633,7 +640,7 @@ func (c *Client) submitVideo(ctx context.Context, client tlsclient.HttpClient, t return respBody, "", fmt.Errorf("%w (%d %s: %s)", ErrAuth, resp.StatusCode, resp.Header.Get("x-access-error"), clip(respBody, 300)) } if resp.StatusCode == 408 || resp.StatusCode == 429 || resp.StatusCode == 451 || resp.StatusCode >= 500 { - return respBody, "", ErrTemporaryUpstream + return respBody, "", ErrDeadUpstream } // "system under load" / timeout_error = adobe overload — treat as a temporary // error so the tempFailover policy moves to the next account (same as the image path). @@ -703,9 +710,12 @@ func (c *Client) pollVideo(ctx context.Context, client tlsclient.HttpClient, tok if resp.StatusCode == 401 || resp.StatusCode == 403 { return nil, nil, fmt.Errorf("%w (%d %s: %s)", ErrAuth, resp.StatusCode, resp.Header.Get("x-access-error"), clip(body, 300)) } - if resp.StatusCode == 429 || resp.StatusCode == 451 || resp.StatusCode >= 500 { + if b := string(body); strings.Contains(b, "system under load") || strings.Contains(b, "timeout_error") { return nil, nil, ErrTemporaryUpstream } + if resp.StatusCode == 429 || resp.StatusCode == 451 || resp.StatusCode >= 500 { + return nil, nil, ErrDeadUpstream + } if resp.StatusCode != 200 { return nil, nil, fmt.Errorf("adobe video poll failed: %d %s", resp.StatusCode, clip(body, 300)) } diff --git a/backend/internal/provider/grok/client.go b/backend/internal/provider/grok/client.go index f625ad8..0dfd6b5 100644 --- a/backend/internal/provider/grok/client.go +++ b/backend/internal/provider/grok/client.go @@ -142,7 +142,9 @@ func (c *Client) FetchCreditsBalance(ctx context.Context, token string) (map[str // GetGrokCreditsConfig field #1 is the credits USED this period (not remaining): // an exhausted account reads 100, a fresh one reads ~0. Remaining = 100 - used. - used, _, ok := parseCreditsConfig(raw) + // Field #5 carries the credits' own reset timestamp (weekly grant refill) — + // this is the 恢复时间 we surface, NOT the subscription's billing-period end. + used, resetUnix, ok := parseCreditsConfig(raw) if !ok { return unknownBalance("unparsable credits config"), nil } @@ -154,15 +156,8 @@ func (c *Client) FetchCreditsBalance(ctx context.Context, token string) (map[str } remaining := fullCredits - used - // 恢复时间: taken solely from the subscription's billing-period end (when the - // plan renews and credits reset). The credits-config weekly reset timestamp is - // intentionally NOT used as a fallback — an account with no active subscription - // has no recovery time. - reset := "" - sub, _ := c.FetchSubscription(ctx, token) - if sub != nil { - reset = strings.TrimSpace(sub.BillingPeriodEnd) - } + // 恢复时间: the credits-config weekly reset (when the free grant refills). + reset := strings.TrimSpace(resetUnix) return map[string]any{ "remaining": remaining, "used": used, @@ -175,7 +170,7 @@ func (c *Client) FetchCreditsBalance(ctx context.Context, token string) (map[str // Subscription is the membership view parsed from GET /rest/subscriptions. type Subscription struct { - Member bool // an active subscription exists + Member bool // a subscription with status ACTIVE exists (INACTIVE entries don't count) Tier string // e.g. SUBSCRIPTION_TIER_GROK_PRO ("" for free) Status string // e.g. SUBSCRIPTION_STATUS_ACTIVE BillingPeriodEnd string // RFC3339; when the plan renews / credits reset @@ -183,7 +178,10 @@ type Subscription struct { } // FetchSubscription reads GET /rest/subscriptions and reports the account's -// membership. An empty subscriptions array means a free account (Member=false). +// membership. Member is true only when an entry with SUBSCRIPTION_STATUS_ACTIVE +// exists: a lapsed membership keeps its entry but flips to +// SUBSCRIPTION_STATUS_INACTIVE, and an empty array means never subscribed — +// both read as Member=false (the entry's tier/status are still surfaced). // A 401/403 maps to ErrAuth; other transport/HTTP errors are returned so callers // can treat them as best-effort (they already have the credit balance). func (c *Client) FetchSubscription(ctx context.Context, token string) (*Subscription, error) { @@ -229,15 +227,17 @@ func (c *Client) FetchSubscription(ctx context.Context, token string) (*Subscrip return nil, fmt.Errorf("%w: subscriptions non-json", ErrTemporaryUpstream) } out := &Subscription{} - // Pick the active subscription (fall back to the first entry) as the membership. + // Surface the ACTIVE subscription if any (falling back to the first entry + // for tier/status info), but only an ACTIVE one sets Member. for i, s := range body.Subscriptions { - if i == 0 || strings.EqualFold(s.Status, "SUBSCRIPTION_STATUS_ACTIVE") { - out.Member = true + active := strings.EqualFold(s.Status, "SUBSCRIPTION_STATUS_ACTIVE") + if i == 0 || active { + out.Member = active out.Tier = strings.TrimSpace(s.Tier) out.Status = strings.TrimSpace(s.Status) out.BillingPeriodEnd = strings.TrimSpace(s.BillingPeriodEnd) out.FreeTrial = s.ActiveOffer.FreeTrial != nil - if strings.EqualFold(s.Status, "SUBSCRIPTION_STATUS_ACTIVE") { + if active { break } } diff --git a/backend/internal/service/maintenance.go b/backend/internal/service/maintenance.go index 02f418a..fde6599 100644 --- a/backend/internal/service/maintenance.go +++ b/backend/internal/service/maintenance.go @@ -131,12 +131,14 @@ func (m *MaintenanceService) tick(ctx context.Context) { log.Printf("maintenance: roll_reset: %v", err) } - // 1b. Runway/grok tokens have no refresh — once the reset marker passes, mark - // them dead directly instead of letting them 401 on next use. Runway's - // marker is the JWT expiry; grok's is the credits reset (grok sso can't be - // renewed either — 失效就失效 — so a purchased short-lived account that has - // lapsed by its reset time is treated as dead rather than re-scheduled). - for _, pool := range []string{"runway", "grok"} { + // 1b. Runway tokens have no refresh — once the JWT expiry marker passes the + // token can only 401, so flip it to disabled+dead proactively instead of + // leaving a doomed account "active". Grok is intentionally NOT swept here: + // its reset marker is billingPeriodEnd (a credits-renewal date), not a + // death deadline — a grok sso keeps working past billingPeriodEnd, so + // expiring on it kills live accounts. Grok death is caught for real by the + // import-time FetchSession check and by marking dead on a 401 at use. + for _, pool := range []string{"runway"} { if n, err := m.tokens.ExpireByReset(ctx, pool); err != nil { log.Printf("maintenance: expire_%s: %v", pool, err) } else if n > 0 { @@ -155,6 +157,10 @@ func (m *MaintenanceService) tick(ctx context.Context) { // always-active account (never went 限额) would otherwise read 0 / 402 // after each reset. Self-guarded + background; no-op once all are done. m.tokenSvc.ActivateKreaDue(ctx) + // 1e. Re-validate grok accounts: an empty /rest/subscriptions (or 401) + // means the membership lapsed → disable+dead; otherwise re-sync the + // credits balance and 恢复时间 (from the credits' weekly reset). + m.tokenSvc.RefreshGrokLiveness(ctx) } // 2. Auto-renew Adobe cookies whose refresh interval has elapsed. diff --git a/backend/internal/service/tokens.go b/backend/internal/service/tokens.go index 162a784..b602ffb 100644 --- a/backend/internal/service/tokens.go +++ b/backend/internal/service/tokens.go @@ -944,6 +944,66 @@ func (s *TokenService) checkPendingGrok(tokenID, ssoToken string) { s.finishPending(ctx, "grok", tokenID, "active", false, quotaMeta) } +// RefreshGrokLiveness re-validates every live grok account each maintenance tick. +// Grok sso can't be renewed and has no reset-based death deadline (billingPeriodEnd +// is only a credits-renewal date — the sso keeps working past it), so liveness is +// probed directly: GET /rest/subscriptions. No ACTIVE entry — a lapsed membership +// flips to SUBSCRIPTION_STATUS_INACTIVE (empty array / 401 also count) — means the +// paid membership is gone → the account is disabled+dead. Otherwise the credits +// balance is re-synced and 恢复时间 is refreshed from the credits' own weekly +// reset (NOT the subscription's billing-period end). +func (s *TokenService) RefreshGrokLiveness(ctx context.Context) { + if s.grok == nil { + return + } + items, err := s.tokens.List(ctx) + if err != nil { + return + } + s.applyProxy(ctx) + for i := range items { + it := items[i] + if it.Pool != "grok" || it.Dead || it.Status == "disabled" || strings.TrimSpace(it.Value) == "" { + continue + } + sub, serr := s.grok.FetchSubscription(ctx, it.Value) + if serr != nil { + if errors.Is(serr, grok.ErrAuth) { + _, _ = s.tokens.Update(ctx, "grok", it.ID, map[string]any{"status": "disabled", "dead": true}) + } + continue // transient upstream error → leave as-is, retry next tick + } + if sub == nil || !sub.Member { + // no ACTIVE subscription (INACTIVE / empty) → membership lapsed → dead. + _, _ = s.tokens.Update(ctx, "grok", it.ID, map[string]any{"status": "disabled", "dead": true}) + continue + } + data, derr := s.grok.FetchCreditsBalance(ctx, it.Value) + if derr != nil { + if errors.Is(derr, grok.ErrAuth) { + _, _ = s.tokens.Update(ctx, "grok", it.ID, map[string]any{"status": "disabled", "dead": true}) + } + continue + } + meta := cloneJSONMap(it.Meta) + meta["cached_quota_at"] = int(time.Now().Unix()) + if rem, ok := data["remaining"].(int); ok { + meta["cached_quota_remaining"] = rem + } + if used, ok := data["used"].(int); ok { + meta["cached_quota_used"] = used + } + if total, ok := data["total"].(int); ok { + meta["cached_quota_total"] = total + } + patch := map[string]any{"meta": meta} + if reset := strings.TrimSpace(stringValue(data["reset_after"])); reset != "" { + patch["cached_quota_reset_after"] = reset + } + _, _ = s.tokens.Update(ctx, "grok", it.ID, patch) + } +} + // ImportCustomAccount adds an upstream as a custom account: base_url + key, the // csv list of model ids it serves (empty = all), plus optional weight and // per-account concurrency. No probe — the account goes active immediately and is @@ -1446,11 +1506,10 @@ func (s *TokenService) Quota(ctx context.Context, pool, id string) (map[string]a meta["cached_quota_total"] = total } patch["meta"] = meta - // Recovery time is the death deadline: the maintenance sweep expires a grok - // account once this marker passes (grok sso can't be renewed). Only stamp it - // when still unset (import couldn't resolve it) — never move it forward on a - // later refresh, so an admin opening 账号管理 can't push the death time out. - if reset := strings.TrimSpace(stringValue(data["reset_after"])); reset != "" && strings.TrimSpace(item.CachedQuotaResetAfter) == "" { + // Recovery time is the credits' weekly reset (when the grant refills) — + // purely informational, NOT a death deadline (liveness is judged by the + // subscriptions sweep / real 401s), so it's safe to refresh every time. + if reset := strings.TrimSpace(stringValue(data["reset_after"])); reset != "" { patch["cached_quota_reset_after"] = reset item.CachedQuotaResetAfter = reset } diff --git a/backend/internal/service/v1.go b/backend/internal/service/v1.go index d315b2b..92aea29 100644 --- a/backend/internal/service/v1.go +++ b/backend/internal/service/v1.go @@ -1303,7 +1303,7 @@ const maxTempDeadAccounts = 3 // auth retry uses a FRESH token instead of replaying the stale one. func (s *V1Service) runPoolWithFailover(ctx context.Context, eventID, pool string, active []model.TokenAccount, kind string, attempt func(token model.TokenAccount) ([]byte, error), - classify func(error) (isAuth, isQuota, isTemporary bool), + classify func(error) (isAuth, isQuota, isTemporary, isDead bool), refreshOnAuth func(tokenID string) (model.TokenAccount, bool), tempFailover bool, ) ([]byte, error) { @@ -1357,7 +1357,7 @@ func (s *V1Service) runPoolWithFailover(ctx context.Context, eventID, pool strin // to the next account. The per-account concurrency gate is held by the caller. func (s *V1Service) tryAccount(ctx context.Context, eventID, pool string, token model.TokenAccount, kind string, attempt func(token model.TokenAccount) ([]byte, error), - classify func(error) (isAuth, isQuota, isTemporary bool), + classify func(error) (isAuth, isQuota, isTemporary, isDead bool), refreshOnAuth func(tokenID string) (model.TokenAccount, bool), tempFailover bool, ) ([]byte, error, bool, bool) { @@ -1375,7 +1375,7 @@ func (s *V1Service) tryAccount(ctx context.Context, eventID, pool string, token }) return data, nil, false, false } - isAuth, isQuota, isTemp := classify(err) + isAuth, isQuota, isTemp, isDead := classify(err) if isQuota { s.markTokenFailure(ctx, pool, token, kind, false, true) return nil, err, true, false @@ -1392,6 +1392,10 @@ 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 { + s.markTokenDead(ctx, pool, token, kind) + return nil, err, true, true + } if isTemp { if tempFailover { // Ops policy (adobe): a temporary upstream error ("system under @@ -1424,8 +1428,8 @@ func (s *V1Service) tryAccount(ctx context.Context, eventID, pool string, token } } -func adobeErrClass(e error) (bool, bool, bool) { - return errors.Is(e, adobe.ErrAuth), errors.Is(e, adobe.ErrQuotaExhausted), errors.Is(e, adobe.ErrTemporaryUpstream) +func adobeErrClass(e error) (bool, bool, bool, bool) { + return errors.Is(e, adobe.ErrAuth), errors.Is(e, adobe.ErrQuotaExhausted), errors.Is(e, adobe.ErrTemporaryUpstream), errors.Is(e, adobe.ErrDeadUpstream) } func (s *V1Service) generateAdobeImage(ctx context.Context, eventID string, modelItem *model.ModelConfig, in V1ImageRequest, aspectRatio, resolution string) ([]byte, error) { @@ -2178,8 +2182,8 @@ func (s *V1Service) generateChatGPTImage(ctx context.Context, eventID string, mo s.reconcileChatGPTQuota(ctx, token.ID, token.Value) } return data, genErr - }, func(e error) (bool, bool, bool) { - return errors.Is(e, chatgpt.ErrAuth), errors.Is(e, chatgpt.ErrQuotaExhausted), errors.Is(e, chatgpt.ErrTemporaryUpstream) + }, func(e error) (bool, bool, bool, bool) { + return errors.Is(e, chatgpt.ErrAuth), errors.Is(e, chatgpt.ErrQuotaExhausted), errors.Is(e, chatgpt.ErrTemporaryUpstream), false }, nil, false) // chatgpt token IS the credential — no cookie to refresh } @@ -2305,8 +2309,8 @@ func (s *V1Service) generateLeonardoImage(ctx context.Context, eventID string, m // sink to 限额 if below the floor (best-effort; never fails a done render). s.reconcileLeonardoCredits(ctx, token.ID, token.Value) return data, nil - }, func(e error) (bool, bool, bool) { - return errors.Is(e, leonardo.ErrAuth), errors.Is(e, leonardo.ErrQuotaExhausted), errors.Is(e, leonardo.ErrTemporaryUpstream) + }, func(e error) (bool, bool, bool, bool) { + return errors.Is(e, leonardo.ErrAuth), errors.Is(e, leonardo.ErrQuotaExhausted), errors.Is(e, leonardo.ErrTemporaryUpstream), false }, nil, false) } @@ -2435,8 +2439,8 @@ func (s *V1Service) generateKreaImage(ctx context.Context, eventID string, model } data, _, genErr := s.krea.GenerateImage(ctx, cookie, in.Prompt, width, height, refs) return data, genErr - }, func(e error) (bool, bool, bool) { - return errors.Is(e, krea.ErrAuth), errors.Is(e, krea.ErrQuotaExhausted), errors.Is(e, krea.ErrTemporaryUpstream) + }, func(e error) (bool, bool, bool, bool) { + return errors.Is(e, krea.ErrAuth), errors.Is(e, krea.ErrQuotaExhausted), errors.Is(e, krea.ErrTemporaryUpstream), false }, nil, false) } @@ -2507,8 +2511,8 @@ func (s *V1Service) generateImagineImage(ctx context.Context, eventID string, mo return nil, genErr } return data, nil - }, func(e error) (bool, bool, bool) { - return errors.Is(e, imagine.ErrAuth), errors.Is(e, imagine.ErrQuotaExhausted), errors.Is(e, imagine.ErrTemporaryUpstream) + }, func(e error) (bool, bool, bool, bool) { + return errors.Is(e, imagine.ErrAuth), errors.Is(e, imagine.ErrQuotaExhausted), errors.Is(e, imagine.ErrTemporaryUpstream), false }, nil, false) } @@ -2955,6 +2959,18 @@ func (s *V1Service) markTokenFailure(ctx context.Context, pool string, token mod _, _ = s.tokens.Update(ctx, pool, token.ID, patch) } +// markTokenDead disables an account and marks it dead on a fatal upstream error +// (a non-overload temporary Adobe failure that ops policy treats as account death). +func (s *V1Service) markTokenDead(ctx context.Context, pool string, token model.TokenAccount, kind string) { + _, _ = s.tokens.Update(ctx, pool, token.ID, map[string]any{ + "last_used_at": time.Now(), + "fail_total": gorm.Expr("fail_total + 1"), + "fails": gorm.Expr("fails + 1"), + "status": "disabled", + "dead": true, + }) +} + // nextCursor returns the pool's current round-robin position and atomically // advances it by one. Concurrent callers each get a distinct value, so parallel // picks land on different accounts instead of racing onto the same one. The