From c0d7c94e2cbe3610ab6de01e76dc027c7a41d6e2 Mon Sep 17 00:00:00 2001 From: Chenyme <118253778+chenyme@users.noreply.github.com> Date: Wed, 19 Aug 2026 16:59:53 +0800 Subject: [PATCH] feat: isolate account-bound proxy leases --- README.md | 2 +- README.zh-CN.md | 2 +- .../internal/application/egress/service.go | 160 ++++++++ .../application/gateway/quality_probe.go | 2 +- .../internal/application/gateway/selector.go | 27 +- .../application/gateway/selector_session.go | 5 + .../application/gateway/selector_test.go | 25 +- .../internal/application/gateway/service.go | 33 +- .../application/gateway/service_test.go | 9 + backend/internal/domain/account/account.go | 30 +- .../relational/account_egress_lease_test.go | 175 +++++++++ .../relational/account_repository.go | 250 ++++++++++++- .../infra/persistence/relational/models.go | 13 + .../infra/persistence/relational/schema.go | 2 + backend/internal/repository/runtime.go | 19 +- .../internal/transport/http/audit/handler.go | 3 +- .../internal/transport/http/egress/handler.go | 162 ++++++++ .../transport/http/egress/handler_test.go | 30 +- .../quality-guard/quality-guard-api.ts | 9 + .../quality-guard/quality-guard-page.tsx | 27 +- frontend/src/shared/i18n/index.ts | 35 +- tools/egress-quality-guard/README.md | 18 + tools/egress-quality-guard/README.zh-CN.md | 10 + tools/egress-quality-guard/quality_guard.py | 354 +++++++++++++++++- .../quality_guard_test.py | 225 ++++++++++- 25 files changed, 1575 insertions(+), 52 deletions(-) create mode 100644 backend/internal/infra/persistence/relational/account_egress_lease_test.go diff --git a/README.md b/README.md index c76ad2767..9a5a1af71 100644 --- a/README.md +++ b/README.md @@ -352,7 +352,7 @@ Egress nodes are scoped to Build, Web, Console, or Web assets. The admin console - Proxy-pool mode without global cooldown after one connection failure - Immediate recovery probes after fixed-proxy transport failures, with per-node coalescing and bounded waiting for fast retry - Optional [Egress Quality Guard](./tools/egress-quality-guard/README.md) for active per-node model probes, guarded quarantine, and recovery; enable it with the built-in `quality-guard` Compose profile -- Give each sticky session its own fixed node (`proxyPool=false`). Do not merge several stickies into one node, or the guard can only quarantine the whole group +- Nodes whose proxy username contains `{account}` are treated as lease-scoped: a passive anomaly temporarily removes only the audited account lease, then recovery pins the probe to that same account and node. An unhealthy probe renews the hold; an expired hold no longer blocks routing if the sidecar is unavailable, so stale guard state cannot strand an account indefinitely. The shared node is never disabled and the rendered proxy identity is never exposed. Ordinary fixed sticky sessions can still be managed as separate nodes Hysteria and TUIC are not supported yet. FlareSolverr accepts only HTTP/SOCKS proxy URLs, so automatic clearance refresh cannot use a tunnel share URL directly. diff --git a/README.zh-CN.md b/README.zh-CN.md index 5987a4b14..584b95b16 100644 --- a/README.zh-CN.md +++ b/README.zh-CN.md @@ -350,7 +350,7 @@ curl http://127.0.0.1:8000/v1/responses \ - 代理池模式，单次连接失败不会触发全局冷却 - 固定代理传输失败后立即复测；同节点复测自动合并，后续绑定请求限时等待并在恢复后快速重试 - 可选的[出口质量守护程序](./tools/egress-quality-guard/README.zh-CN.md)，支持逐节点模型探测、防误杀隔离和自动恢复；通过内置的 `quality-guard` Compose profile 按需启用 -- 固定 sticky 会话应各自建成独立节点（`proxyPool=false`）。不要把多条 sticky 合成一个节点，否则质量守护只能整组摘流，无法定位坏会话 +- 代理用户名包含 `{account}` 的节点会被识别为租约级节点：被动审计异常只会临时移出对应的账号租约，冷却后固定使用同一账号和节点复测，复测异常会续期；若 sidecar 不可用，已到期的隔离不会继续阻断路由，避免孤儿状态永久卡住账号。共享节点始终不会因此停用，也不会暴露渲染后的代理身份。普通固定 sticky 会话仍可按独立节点管理 Hysteria 与 TUIC 暂未支持。FlareSolverr 仅接受 HTTP/SOCKS 代理地址，因此自动刷新 Clearance 暂不能直接使用隧道分享链接。 diff --git a/backend/internal/application/egress/service.go b/backend/internal/application/egress/service.go index 481ead4f7..8e1f3531f 100644 --- a/backend/internal/application/egress/service.go +++ b/backend/internal/application/egress/service.go @@ -25,6 +25,8 @@ var ( ErrProbeStale = errors.New("代理配置在探测期间已更新，请重新测试") ErrQualityProbeUnavailable = errors.New("出口质量探测不可用") ErrQualityProbeNoAccount = errors.New("质量检测暂无可调度账号") + ErrQualityLeaseUnavailable = errors.New("租约级质量隔离不可用") + ErrQualityLeaseConflict = errors.New("租约状态已变化") ErrClearanceUnavailable = errors.New("Clearance 刷新不可用") ErrProxyProfileUnavailable = errors.New("共享代理配置功能不可用") ErrProxyProfileInUse = errors.New("共享代理配置仍被节点使用") @@ -42,6 +44,7 @@ const ( type QualityProbeInput struct { ClientKeyID uint64 + AccountID uint64 Model string Prompt string Expected string @@ -116,6 +119,7 @@ type Service struct { repository ServiceRepository proxyProfiles repository.EgressProxyProfileRepository accounts AccountBindingRepository + qualityLeases QualityLeaseRepository operations OperationsRepository cipher *security.Cipher mu sync.RWMutex @@ -131,6 +135,25 @@ type Service struct { autoAssignMaxMigrationShare float64 } +// QualityLeaseRepository is optional and deliberately separate from account +// binding administration. It exposes only the state required to isolate one +// account-bound proxy lease. +type QualityLeaseRepository interface { + Get(context.Context, uint64) (accountdomain.Credential, error) + ListEgressLeaseBlocks(context.Context, int, *accountdomain.EgressLeaseBlockCursor) ([]accountdomain.EgressLeaseBlock, error) + UpsertEgressLeaseBlock(context.Context, accountdomain.EgressLeaseBlock) (accountdomain.EgressLeaseBlock, error) + DeleteEgressLeaseBlock(context.Context, uint64, uint64, string) (bool, error) + DeleteEgressLeaseBlocksByNodes(context.Context, []uint64) (int64, error) + PruneInvalidEgressLeaseBlocks(context.Context, int) (int64, error) +} + +type QualityLeaseInput struct { + AccountID uint64 + NodeID uint64 + Reason string + QuarantineSeconds int +} + func (s *Service) SetQualityProber(value QualityProber) { s.mu.Lock() s.qualityProber = value @@ -174,6 +197,15 @@ func (s *Service) ProbeQuality(ctx context.Context, nodeID uint64, input Quality if node.Scope != domain.ScopeBuild || strings.TrimSpace(node.EncryptedProxyURL) == "" { return QualityProbeResult{}, fmt.Errorf("%w: 质量探测仅支持已配置代理的 grok_build 节点", ErrInvalidInput) } + if input.AccountID != 0 { + if !s.accountBoundProxy(node) || s.qualityLeases == nil { + return QualityProbeResult{}, fmt.Errorf("%w: 账号定向探测仅支持按账号派生代理的节点", ErrInvalidInput) + } + credential, loadErr := s.qualityLeases.Get(ctx, input.AccountID) + if loadErr != nil || credential.Provider != accountdomain.ProviderBuild || !credential.Enabled || credential.AuthStatus != accountdomain.AuthStatusActive || credential.EgressNodeID != nodeID { + return QualityProbeResult{}, ErrQualityProbeNoAccount + } + } s.mu.RLock() prober := s.qualityProber s.mu.RUnlock() @@ -241,10 +273,111 @@ func NewService(storage ServiceRepository, cipher *security.Cipher, browserUA st } if len(accounts) > 0 { service.accounts = accounts[0] + if leases, ok := accounts[0].(QualityLeaseRepository); ok { + service.qualityLeases = leases + } } return service } +func (s *Service) ListQualityLeases(ctx context.Context, limit int, cursor *accountdomain.EgressLeaseBlockCursor) ([]accountdomain.EgressLeaseBlock, error) { + if s.qualityLeases == nil { + return nil, ErrQualityLeaseUnavailable + } + if limit < 1 || limit > 1001 { + return nil, ErrInvalidInput + } + if cursor == nil { + for range 10 { + pruned, err := s.qualityLeases.PruneInvalidEgressLeaseBlocks(ctx, 1000) + if err != nil { + return nil, err + } + if pruned < 1000 { + break + } + } + nodes, err := s.repository.ListEgressNodes(ctx, domain.ScopeBuild, repository.SortQuery{}) + if err != nil { + return nil, err + } + staleNodeIDs := make([]uint64, 0) + for _, node := range nodes { + if !node.Enabled || !s.accountBoundProxy(node) { + staleNodeIDs = append(staleNodeIDs, node.ID) + } + } + if _, err := s.qualityLeases.DeleteEgressLeaseBlocksByNodes(ctx, staleNodeIDs); err != nil { + return nil, err + } + } + return s.qualityLeases.ListEgressLeaseBlocks(ctx, limit, cursor) +} + +func (s *Service) QuarantineQualityLease(ctx context.Context, input QualityLeaseInput) (accountdomain.EgressLeaseBlock, error) { + input.Reason = strings.TrimSpace(input.Reason) + if input.AccountID == 0 || input.NodeID == 0 || input.QuarantineSeconds < 30 || input.QuarantineSeconds > 86400 || !qualityLeaseReasonAllowed(input.Reason) { + return accountdomain.EgressLeaseBlock{}, ErrInvalidInput + } + if s.qualityLeases == nil { + return accountdomain.EgressLeaseBlock{}, ErrQualityLeaseUnavailable + } + node, err := s.repository.GetEgressNode(ctx, input.NodeID) + if errors.Is(err, repository.ErrNotFound) { + return accountdomain.EgressLeaseBlock{}, ErrNotFound + } + if err != nil { + return accountdomain.EgressLeaseBlock{}, err + } + if !node.Enabled || node.Scope != domain.ScopeBuild || !s.accountBoundProxy(node) { + return accountdomain.EgressLeaseBlock{}, ErrInvalidInput + } + credential, err := s.qualityLeases.Get(ctx, input.AccountID) + if err != nil || credential.Provider != accountdomain.ProviderBuild || !credential.Enabled || credential.AuthStatus != accountdomain.AuthStatusActive || credential.EgressNodeID != input.NodeID { + return accountdomain.EgressLeaseBlock{}, ErrQualityLeaseConflict + } + version, err := security.NewOpaqueToken(18) + if err != nil { + return accountdomain.EgressLeaseBlock{}, err + } + now := time.Now().UTC() + value, err := s.qualityLeases.UpsertEgressLeaseBlock(ctx, accountdomain.EgressLeaseBlock{ + AccountID: input.AccountID, NodeID: input.NodeID, Reason: input.Reason, Version: version, + CooldownUntil: now.Add(time.Duration(input.QuarantineSeconds) * time.Second), UpdatedAt: now, + }) + if errors.Is(err, repository.ErrConflict) || errors.Is(err, repository.ErrNotFound) { + return accountdomain.EgressLeaseBlock{}, ErrQualityLeaseConflict + } + return value, err +} + +func (s *Service) RestoreQualityLease(ctx context.Context, accountID, nodeID uint64, version string) (bool, error) { + version = strings.TrimSpace(version) + if accountID == 0 || nodeID == 0 || version == "" || len(version) > 64 { + return false, ErrInvalidInput + } + if s.qualityLeases == nil { + return false, ErrQualityLeaseUnavailable + } + restored, err := s.qualityLeases.DeleteEgressLeaseBlock(ctx, accountID, nodeID, version) + if err != nil { + return false, err + } + if !restored { + return false, ErrQualityLeaseConflict + } + return true, nil +} + +func qualityLeaseReasonAllowed(value string) bool { + switch value { + case "hard_tps", "soft_tps", "buffered_burst", "missing_thinking", "expected_marker_missing", "insufficient_output_tokens", "insufficient_generation_window", "probe_errors", "recovery_probe_error", "rotation_error": + return true + default: + return false + } +} + func (s *Service) UpdateDefaults(browserUA string) { s.mu.Lock() defer s.mu.Unlock() @@ -368,6 +501,7 @@ func (s *Service) Update(ctx context.Context, id uint64, input Input) (domain.Pu return domain.PublicNode{}, err } previousScope := value.Scope + previousProxyURL := value.EncryptedProxyURL value, err = s.applyInput(value, input, false) if err != nil { return domain.PublicNode{}, err @@ -386,6 +520,11 @@ func (s *Service) Update(ctx context.Context, id uint64, input Input) (domain.Pu } if err == nil { s.forgetClearance(updated.ID) + if !updated.Enabled || previousScope != updated.Scope || previousProxyURL != updated.EncryptedProxyURL { + if clearErr := s.clearQualityLeasesForNodes(ctx, []uint64{updated.ID}); clearErr != nil { + return s.publicNode(updated), clearErr + } + } } return s.publicNode(updated), err } @@ -503,6 +642,9 @@ func (s *Service) UpdateProxyProfile(ctx context.Context, id uint64, input Proxy } if proxyChanged { s.forgetClearances(nodeIDs) + if err := s.clearQualityLeasesForNodes(ctx, nodeIDs); err != nil { + return s.publicProxyProfile(updated), err + } } return s.publicProxyProfile(updated), nil } @@ -695,6 +837,11 @@ func (s *Service) UpdateManyEnabled(ctx context.Context, nodeIDs []uint64, enabl } if updated > 0 { s.forgetClearances(ids) + if !enabled { + if clearErr := s.clearQualityLeasesForNodes(ctx, ids); clearErr != nil { + return updated, clearErr + } + } } return updated, nil } @@ -718,9 +865,22 @@ func (s *Service) UpdateManyEnabled(ctx context.Context, nodeIDs []uint64, enabl s.forgetClearance(id) updated++ } + if !enabled && updated > 0 { + if err := s.clearQualityLeasesForNodes(ctx, ids); err != nil { + return updated, err + } + } return updated, nil } +func (s *Service) clearQualityLeasesForNodes(ctx context.Context, nodeIDs []uint64) error { + if s.qualityLeases == nil || len(nodeIDs) == 0 { + return nil + } + _, err := s.qualityLeases.DeleteEgressLeaseBlocksByNodes(ctx, uniqueIDs(nodeIDs)) + return err +} + func (s *Service) Delete(ctx context.Context, id uint64) error { err := s.repository.DeleteEgressNode(ctx, id) if errors.Is(err, repository.ErrNotFound) { diff --git a/backend/internal/application/gateway/quality_probe.go b/backend/internal/application/gateway/quality_probe.go index e66870afd..c796bfe33 100644 --- a/backend/internal/application/gateway/quality_probe.go +++ b/backend/internal/application/gateway/quality_probe.go @@ -78,7 +78,7 @@ func (s *Service) ProbeEgressQuality(ctx context.Context, nodeID uint64, input e probeCtx := infraegress.WithQualityProbe(ctx) result, err := s.CreateChatCompletion(probeCtx, Input{ RequestID: requestID, ClientKey: key, PublicModel: publicModel, Body: body, - Streaming: true, Operation: audit.OperationChat, ForcedEgressNodeID: nodeID, + Streaming: true, Operation: audit.OperationChat, ForcedEgressNodeID: nodeID, ForcedAccountID: input.AccountID, }) if err != nil { return egressapp.QualityProbeResult{}, normalizeQualityProbeRequestError(err) diff --git a/backend/internal/application/gateway/selector.go b/backend/internal/application/gateway/selector.go index 32874d160..db0fe95ca 100644 --- a/backend/internal/application/gateway/selector.go +++ b/backend/internal/application/gateway/selector.go @@ -491,6 +491,11 @@ func (s *Selector) acquire(ctx context.Context, provider account.Provider, model earliestRetry = earlierFuture(earliestRetry, candidate.ModelQuotaBlock.CooldownUntil, now) continue } + if candidateEgressLeaseCooling(candidate, value, now) { + coolingCandidates++ + earliestRetry = earlierFuture(earliestRetry, candidate.EgressLeaseBlock.CooldownUntil, now) + continue + } if value.CooldownUntil != nil && now.Before(*value.CooldownUntil) { coolingCandidates++ earliestRetry = earlierFuture(earliestRetry, *value.CooldownUntil, now) @@ -752,14 +757,20 @@ func isSelectionUnavailable(err error, reason SelectionUnavailableReason) bool { // AcquirePinned 为 previous_response_id 等账号归属请求获取指定账号租约。 func (s *Selector) AcquirePinned(ctx context.Context, provider account.Provider, accountID, modelRouteID uint64, upstreamModel, quotaMode string, inference bool) (*accountLease, error) { - return s.acquirePinned(ctx, provider, accountID, modelRouteID, upstreamModel, quotaMode, inference, clientkeydomain.AccountScope{}) + return s.acquirePinned(ctx, provider, accountID, modelRouteID, upstreamModel, quotaMode, inference, false, clientkeydomain.AccountScope{}) } func (s *Selector) AcquirePinnedForKey(ctx context.Context, provider account.Provider, accountID, modelRouteID uint64, upstreamModel, quotaMode string, inference bool, scope clientkeydomain.AccountScope) (*accountLease, error) { - return s.acquirePinned(ctx, provider, accountID, modelRouteID, upstreamModel, quotaMode, inference, scope) + return s.acquirePinned(ctx, provider, accountID, modelRouteID, upstreamModel, quotaMode, inference, false, scope) +} + +// AcquirePinnedForQualityProbe keeps every ordinary eligibility check while +// bypassing only the exact account+node lease block being recovery-tested. +func (s *Selector) AcquirePinnedForQualityProbe(ctx context.Context, provider account.Provider, accountID, modelRouteID uint64, upstreamModel, quotaMode string, scope clientkeydomain.AccountScope) (*accountLease, error) { + return s.acquirePinned(ctx, provider, accountID, modelRouteID, upstreamModel, quotaMode, true, true, scope) } -func (s *Selector) acquirePinned(ctx context.Context, provider account.Provider, accountID, modelRouteID uint64, upstreamModel, quotaMode string, inference bool, requestedScope clientkeydomain.AccountScope) (lease *accountLease, err error) { +func (s *Selector) acquirePinned(ctx context.Context, provider account.Provider, accountID, modelRouteID uint64, upstreamModel, quotaMode string, inference, ignoreEgressLeaseBlock bool, requestedScope clientkeydomain.AccountScope) (lease *accountLease, err error) { accountScope, scopeValid := clientkeydomain.NormalizeAccountScope(requestedScope) defer annotateSelectionAccountScope(&err, accountScope) if !scopeValid || !accountScope.AllowsProvider(provider) { @@ -790,6 +801,9 @@ func (s *Selector) acquirePinned(ctx context.Context, provider account.Provider, if candidate.ModelQuotaBlock != nil && now.Before(candidate.ModelQuotaBlock.CooldownUntil) { return nil, &SelectionUnavailableError{Reason: SelectionModelCooling, RetryAfter: retryDelay(now, candidate.ModelQuotaBlock.CooldownUntil)} } + if !ignoreEgressLeaseBlock && candidateEgressLeaseCooling(candidate, value, now) { + return nil, &SelectionUnavailableError{Reason: SelectionCooling, RetryAfter: retryDelay(now, candidate.EgressLeaseBlock.CooldownUntil)} + } if value.CooldownUntil != nil && now.Before(*value.CooldownUntil) { return nil, &SelectionUnavailableError{Reason: SelectionCooling, RetryAfter: retryDelay(now, *value.CooldownUntil)} } @@ -901,6 +915,11 @@ func effectiveQuotaMode(candidate account.RoutingCandidate, fallback string) str return fallback } +func candidateEgressLeaseCooling(candidate account.RoutingCandidate, credential account.Credential, now time.Time) bool { + block := candidate.EgressLeaseBlock + return block != nil && block.AccountID == credential.ID && block.NodeID != 0 && credential.EgressNodeID == block.NodeID && now.Before(block.CooldownUntil) +} + // candidateSupportsModel treats a recognized Web catalog entry as an // effective capability for tiers that the adapter explicitly allows. This // prevents a historical capability snapshot from blocking a newly enabled @@ -1889,7 +1908,7 @@ func assembleRoutingCandidates(provider account.Provider, quotaMode string, base } result = append(result, account.RoutingCandidate{ Credential: base.Credential, Billing: base.Billing, QuotaWindow: base.QuotaWindow, QuotaRecovery: base.QuotaRecovery, - ModelQuotaBlock: overlayValue.ModelQuotaBlock, ModelCapabilityKnown: known, SupportsModel: supports, + EgressLeaseBlock: base.EgressLeaseBlock, ModelQuotaBlock: overlayValue.ModelQuotaBlock, ModelCapabilityKnown: known, SupportsModel: supports, }) } return result diff --git a/backend/internal/application/gateway/selector_session.go b/backend/internal/application/gateway/selector_session.go index 37682d894..7cbbce849 100644 --- a/backend/internal/application/gateway/selector_session.go +++ b/backend/internal/application/gateway/selector_session.go @@ -84,6 +84,11 @@ func (s *Selector) beginSelectionSessionForKey(ctx context.Context, provider acc earliestRetry = earlierFuture(earliestRetry, candidate.ModelQuotaBlock.CooldownUntil, now) continue } + if candidateEgressLeaseCooling(candidate, value, now) { + coolingCandidates++ + earliestRetry = earlierFuture(earliestRetry, candidate.EgressLeaseBlock.CooldownUntil, now) + continue + } if value.CooldownUntil != nil && now.Before(*value.CooldownUntil) { coolingCandidates++ earliestRetry = earlierFuture(earliestRetry, *value.CooldownUntil, now) diff --git a/backend/internal/application/gateway/selector_test.go b/backend/internal/application/gateway/selector_test.go index fef73bd53..ea2e267de 100644 --- a/backend/internal/application/gateway/selector_test.go +++ b/backend/internal/application/gateway/selector_test.go @@ -141,10 +141,33 @@ func TestSelectorQualityProbePinsAccountToRequestedEgressNode(t *testing.T) { if err != nil { t.Fatal(err) } - defer lease.Release() if lease.Credential.ID != second.ID || lease.Credential.ID == first.ID { t.Fatalf("selected account=%d, want=%d on node=%d", lease.Credential.ID, second.ID, secondNode.ID) } + lease.Release() + + if _, err := accounts.UpsertEgressLeaseBlock(ctx, account.EgressLeaseBlock{ + AccountID: second.ID, NodeID: secondNode.ID, Reason: "hard_tps", Version: "selector-lease-0001", CooldownUntil: time.Now().UTC().Add(time.Hour), + }); err != nil { + t.Fatal(err) + } + selector = NewSelector(accounts, memory.NewConcurrencyLimiter(), memory.NewStickyStore(), nil, time.Hour, time.Second, time.Minute) + if _, err := selector.AcquirePinnedForKey(ctx, account.ProviderBuild, second.ID, 0, "grok-test", "", true, clientkeydomain.AccountScope{}); err == nil { + t.Fatal("ordinary pinned inference ignored the active egress lease block") + } else { + var unavailable *SelectionUnavailableError + if !errors.As(err, &unavailable) || unavailable.Reason != SelectionCooling { + t.Fatalf("ordinary pinned error = %v", err) + } + } + recoveryLease, err := selector.AcquirePinnedForQualityProbe(ctx, account.ProviderBuild, second.ID, 0, "grok-test", "", clientkeydomain.AccountScope{}) + if err != nil { + t.Fatalf("quality recovery did not bypass only the lease block: %v", err) + } + defer recoveryLease.Release() + if recoveryLease.Credential.ID != second.ID || recoveryLease.Credential.EgressNodeID != secondNode.ID { + t.Fatalf("quality recovery lease = %#v", recoveryLease.Credential) + } } func TestSelectorQualityProbeBorrowsHealthyAccountForUnavailableNode(t *testing.T) { diff --git a/backend/internal/application/gateway/service.go b/backend/internal/application/gateway/service.go index b4b0136f3..129b78115 100644 --- a/backend/internal/application/gateway/service.go +++ b/backend/internal/application/gateway/service.go @@ -74,6 +74,13 @@ func newRoutingAttemptPolicy(configured int) routingAttemptPolicy { return routingAttemptPolicy{limit: configured} } +func newRequestRoutingAttemptPolicy(configured int, pinned bool) routingAttemptPolicy { + if pinned { + return newRoutingAttemptPolicy(1) + } + return newRoutingAttemptPolicy(configured) +} + func (p routingAttemptPolicy) allows(attempt int) bool { return p.unlimited || attempt < p.limit } @@ -116,6 +123,9 @@ type Input struct { // ForcedEgressNodeID is an internal-only administrator probe constraint. // Public inference handlers never populate it. ForcedEgressNodeID uint64 + // ForcedAccountID is paired with ForcedEgressNodeID only by the internal + // Quality Guard recovery path. It never accepts public request input. + ForcedAccountID uint64 } type Usage struct { @@ -1001,11 +1011,11 @@ func (s *Service) createResponseAt(ctx context.Context, input Input, path string if input.PreviousResponseID != "" && !supportsStoredResponses { return nil, ErrResponseStateUnsupported } - attemptPolicy := newRoutingAttemptPolicy(int(s.maxAttempts.Load())) + // A lease recovery probe stays on exactly one account and one rendered proxy + // identity. Retrying the same pinned account would provide neither failover + // nor new evidence and can multiply a slow/failing probe. + attemptPolicy := newRequestRoutingAttemptPolicy(int(s.maxAttempts.Load()), ownership != nil || input.ForcedAccountID != 0) idempotencyID, _ := security.NewOpaqueToken(18) - if ownership != nil { - attemptPolicy = newRoutingAttemptPolicy(1) - } pricingModel := s.providers.PricingModel(route.Provider, route.UpstreamModel) if err := s.checkLedgerReady(); err != nil { return nil, err @@ -1197,7 +1207,18 @@ attemptLoop: var lease *accountLease var err error selectionStarted := time.Now() - if ownership != nil { + if input.ForcedAccountID != 0 { + if input.ForcedEgressNodeID == 0 { + err = &SelectionUnavailableError{Reason: SelectionNoAccounts} + } else { + lease, err = s.selector.AcquirePinnedForQualityProbe(ctx, route.Provider, input.ForcedAccountID, route.ID, route.UpstreamModel, quotaMode, accountScope) + if err == nil && lease.Credential.EgressNodeID != input.ForcedEgressNodeID { + lease.Release() + lease = nil + err = &SelectionUnavailableError{Reason: SelectionNoAccounts} + } + } + } else if ownership != nil { lease, err = s.selector.AcquirePinnedForKey(ctx, route.Provider, ownership.AccountID, route.ID, route.UpstreamModel, quotaMode, true, accountScope) } else if input.ForcedEgressNodeID != 0 { lease, err = s.selector.AcquireForKeyOnEgressNode(ctx, route.Provider, route.ID, route.UpstreamModel, quotaMode, affinityKey, excluded, !quotaProbeAttempted, accountScope, input.ForcedEgressNodeID) @@ -1229,7 +1250,7 @@ attemptLoop: // Stored Responses are pinned to one account. Return the cached 429 // immediately instead of spinning until the cooldown expires or // replaying the request on the same account. - if ownership != nil { + if ownership != nil || input.ForcedAccountID != 0 { break attemptLoop } attempt-- diff --git a/backend/internal/application/gateway/service_test.go b/backend/internal/application/gateway/service_test.go index 0f35a8335..f493abc5d 100644 --- a/backend/internal/application/gateway/service_test.go +++ b/backend/internal/application/gateway/service_test.go @@ -555,6 +555,15 @@ func TestRoutingAttemptPolicy(t *testing.T) { } } +func TestPinnedRequestAttemptPolicyAlwaysAllowsOneAttempt(t *testing.T) { + for _, configured := range []int{1, 6, unlimitedRoutingAttempts} { + policy := newRequestRoutingAttemptPolicy(configured, true) + if !policy.allows(0) || policy.allows(1) || policy.hasNext(0) { + t.Fatalf("configured=%d pinned policy = %#v", configured, policy) + } + } +} + func TestGatewayUnlimitedAttemptsExhaustsEligiblePool(t *testing.T) { ctx := context.Background() database, err := relational.OpenSQLite(ctx, filepath.Join(t.TempDir(), "gateway-unlimited-attempts.db")) diff --git a/backend/internal/domain/account/account.go b/backend/internal/domain/account/account.go index e4495b7e0..cfd372291 100644 --- a/backend/internal/domain/account/account.go +++ b/backend/internal/domain/account/account.go @@ -443,6 +443,7 @@ type RoutingCandidate struct { Billing *Billing QuotaWindow *QuotaWindow QuotaRecovery *QuotaRecovery + EgressLeaseBlock *EgressLeaseBlock ModelQuotaBlock *ModelQuotaBlock ModelCapabilityKnown bool SupportsModel bool @@ -451,10 +452,11 @@ type RoutingCandidate struct { // RoutingAccountBase contains provider-level routing state reusable across // models. Credential material is hydrated only after an account is selected. type RoutingAccountBase struct { - Credential Credential - Billing *Billing - QuotaRecovery *QuotaRecovery - QuotaWindow *QuotaWindow + Credential Credential + Billing *Billing + QuotaRecovery *QuotaRecovery + QuotaWindow *QuotaWindow + EgressLeaseBlock *EgressLeaseBlock } // RoutingAccountOverlay contains model-specific eligibility state. @@ -480,6 +482,26 @@ type ModelQuotaBlock struct { UpdatedAt time.Time } +// EgressLeaseBlock temporarily removes one account-bound proxy lease from +// routing without changing the account's health or disabling the physical +// egress node shared by other leases. +type EgressLeaseBlock struct { + AccountID uint64 + NodeID uint64 + Reason string + Version string + CooldownUntil time.Time + UpdatedAt time.Time +} + +// EgressLeaseBlockCursor is the stable keyset position used to scan durable +// lease state while rows may be renewed or removed concurrently. +type EgressLeaseBlockCursor struct { + CooldownUntil time.Time + AccountID uint64 + NodeID uint64 +} + // DeviceSession 表示一次短期 Device OAuth 授权流程。 type DeviceSession struct { ID string diff --git a/backend/internal/infra/persistence/relational/account_egress_lease_test.go b/backend/internal/infra/persistence/relational/account_egress_lease_test.go new file mode 100644 index 000000000..6ba4a7415 --- /dev/null +++ b/backend/internal/infra/persistence/relational/account_egress_lease_test.go @@ -0,0 +1,175 @@ +package relational + +import ( + "context" + "path/filepath" + "testing" + "time" + + "github.com/chenyme/grok2api/backend/internal/domain/account" + "github.com/chenyme/grok2api/backend/internal/repository" +) + +func TestEgressLeaseBlockRoutesOnlyMatchingActiveBindingAndUsesCAS(t *testing.T) { + ctx := context.Background() + database, err := OpenSQLite(ctx, filepath.Join(t.TempDir(), "egress-lease.db")) + if err != nil { + t.Fatal(err) + } + defer database.Close() + if err := database.InitializeSchema(ctx); err != nil { + t.Fatal(err) + } + accounts := NewAccountRepository(database) + nodes := NewEgressRepository(database) + cipher := egressOperationsCipher(t) + firstNode := createHealthyEgressNode(t, ctx, nodes, cipher, "lease-first", 0) + secondNode := createHealthyEgressNode(t, ctx, nodes, cipher, "lease-second", 0) + credential := createEgressOperationsAccount(t, ctx, accounts, "lease-account") + if _, err := accounts.UpdateEgressBindings(ctx, account.ProviderBuild, []uint64{credential.ID}, &firstNode.ID, account.EgressAssignmentManual, time.Now().UTC()); err != nil { + t.Fatal(err) + } + + until := time.Now().UTC().Add(time.Hour) + stored, err := accounts.UpsertEgressLeaseBlock(ctx, account.EgressLeaseBlock{ + AccountID: credential.ID, NodeID: firstNode.ID, Reason: "hard_tps", Version: "lease-version-0001", CooldownUntil: until, + }) + if err != nil { + t.Fatal(err) + } + if stored.Version != "lease-version-0001" { + t.Fatalf("stored block = %#v", stored) + } + candidates, err := accounts.ListRoutingCandidates(ctx, account.ProviderBuild, 0, "grok-test", "") + if err != nil { + t.Fatal(err) + } + if len(candidates) != 1 || candidates[0].EgressLeaseBlock == nil || candidates[0].EgressLeaseBlock.NodeID != firstNode.ID { + t.Fatalf("routing candidates = %#v", candidates) + } + + shorter, err := accounts.UpsertEgressLeaseBlock(ctx, account.EgressLeaseBlock{ + AccountID: credential.ID, NodeID: firstNode.ID, Reason: "soft_tps", Version: "lease-version-0002", CooldownUntil: until.Add(-time.Minute), + }) + if err != nil { + t.Fatal(err) + } + if shorter.Version != stored.Version || !shorter.CooldownUntil.Equal(stored.CooldownUntil) { + t.Fatalf("shorter hold replaced stronger block: %#v", shorter) + } + if deleted, err := accounts.DeleteEgressLeaseBlock(ctx, credential.ID, firstNode.ID, "lease-version-stale"); err != nil || deleted { + t.Fatalf("stale CAS delete = %v, %v", deleted, err) + } + + if _, err := accounts.UpdateEgressBindings(ctx, account.ProviderBuild, []uint64{credential.ID}, &secondNode.ID, account.EgressAssignmentManual, time.Now().UTC()); err != nil { + t.Fatal(err) + } + blocks, err := accounts.ListEgressLeaseBlocks(ctx, 10, nil) + if err != nil { + t.Fatal(err) + } + if len(blocks) != 0 { + t.Fatalf("rebinding left stale lease blocks: %#v", blocks) + } +} + +func TestExpiredEgressLeaseBlockIsReconciledButNotRouted(t *testing.T) { + ctx := context.Background() + database, err := OpenSQLite(ctx, filepath.Join(t.TempDir(), "expired-egress-lease.db")) + if err != nil { + t.Fatal(err) + } + defer database.Close() + if err := database.InitializeSchema(ctx); err != nil { + t.Fatal(err) + } + accounts := NewAccountRepository(database) + nodes := NewEgressRepository(database) + node := createHealthyEgressNode(t, ctx, nodes, egressOperationsCipher(t), "lease-expired", 0) + credential := createEgressOperationsAccount(t, ctx, accounts, "lease-expired-account") + if _, err := accounts.UpdateEgressBindings(ctx, account.ProviderBuild, []uint64{credential.ID}, &node.ID, account.EgressAssignmentManual, time.Now().UTC()); err != nil { + t.Fatal(err) + } + if _, err := accounts.UpsertEgressLeaseBlock(ctx, account.EgressLeaseBlock{ + AccountID: credential.ID, NodeID: node.ID, Reason: "hard_tps", Version: "lease-version-0001", CooldownUntil: time.Now().UTC().Add(-time.Second), + }); err != nil { + t.Fatal(err) + } + blocks, err := accounts.ListEgressLeaseBlocks(ctx, 10, nil) + if err != nil || len(blocks) != 1 { + t.Fatalf("reconciliation blocks = %#v, %v", blocks, err) + } + candidates, err := accounts.ListRoutingCandidates(ctx, account.ProviderBuild, 0, "grok-test", "") + if err != nil { + t.Fatal(err) + } + if len(candidates) != 1 || candidates[0].EgressLeaseBlock != nil { + t.Fatalf("expired block remained routable: %#v", candidates) + } +} + +func TestEgressLeaseBlockKeysetPaginationAndInvalidPruning(t *testing.T) { + ctx := context.Background() + database, err := OpenSQLite(ctx, filepath.Join(t.TempDir(), "egress-lease-page.db")) + if err != nil { + t.Fatal(err) + } + defer database.Close() + if err := database.InitializeSchema(ctx); err != nil { + t.Fatal(err) + } + accounts := NewAccountRepository(database) + nodes := NewEgressRepository(database) + node := createHealthyEgressNode(t, ctx, nodes, egressOperationsCipher(t), "lease-page", 0) + base := time.Now().UTC().Add(time.Hour) + created := make([]account.Credential, 0, 5) + for index := range 5 { + credential := createEgressOperationsAccount(t, ctx, accounts, "lease-page-account-"+string(rune('a'+index))) + created = append(created, credential) + if _, err := accounts.UpdateEgressBindings(ctx, account.ProviderBuild, []uint64{credential.ID}, &node.ID, account.EgressAssignmentManual, time.Now().UTC()); err != nil { + t.Fatal(err) + } + if _, err := accounts.UpsertEgressLeaseBlock(ctx, account.EgressLeaseBlock{ + AccountID: credential.ID, NodeID: node.ID, Reason: "hard_tps", Version: "lease-page-version-" + string(rune('a'+index)), CooldownUntil: base.Add(time.Duration(index) * time.Second), + }); err != nil { + t.Fatal(err) + } + } + first, err := accounts.ListEgressLeaseBlocks(ctx, 2, nil) + if err != nil || len(first) != 2 { + t.Fatalf("first page = %#v, %v", first, err) + } + cursor := &account.EgressLeaseBlockCursor{CooldownUntil: first[1].CooldownUntil, AccountID: first[1].AccountID, NodeID: first[1].NodeID} + second, err := accounts.ListEgressLeaseBlocks(ctx, 2, cursor) + if err != nil || len(second) != 2 || second[0].AccountID == first[1].AccountID { + t.Fatalf("second page = %#v, %v", second, err) + } + if _, err := accounts.UpdateMany(ctx, account.ProviderBuild, []uint64{created[0].ID}, repository.AccountUpdates{Enabled: boolPointer(false)}); err != nil { + t.Fatal(err) + } + remaining, err := accounts.ListEgressLeaseBlocks(ctx, 10, nil) + if err != nil || len(remaining) != 4 { + t.Fatalf("remaining blocks = %#v, %v", remaining, err) + } + reauth := created[1] + reauth.AuthStatus = account.AuthStatusReauthRequired + if _, err := accounts.Update(ctx, reauth); err != nil { + t.Fatal(err) + } + remaining, err = accounts.ListEgressLeaseBlocks(ctx, 10, nil) + if err != nil || len(remaining) != 3 { + t.Fatalf("reauth cleanup blocks = %#v, %v", remaining, err) + } + + disabled := node + disabled.Enabled = false + if _, err := nodes.UpdateEgressNode(ctx, disabled); err != nil { + t.Fatal(err) + } + pruned, err := accounts.PruneInvalidEgressLeaseBlocks(ctx, 1000) + if err != nil || pruned != 3 { + t.Fatalf("pruned = %d, %v", pruned, err) + } +} + +func boolPointer(value bool) *bool { return &value } diff --git a/backend/internal/infra/persistence/relational/account_repository.go b/backend/internal/infra/persistence/relational/account_repository.go index d6f520877..faf6104e0 100644 --- a/backend/internal/infra/persistence/relational/account_repository.go +++ b/backend/internal/infra/persistence/relational/account_repository.go @@ -366,6 +366,10 @@ func (r *AccountRepository) ListRoutingCandidates(ctx context.Context, provider if err != nil { return nil, err } + egressLeaseBlocks, err := r.getRoutingEgressLeaseBlocks(ctx, provider, values, time.Now().UTC()) + if err != nil { + return nil, err + } known := make(map[uint64]bool, len(values)) supported := make(map[uint64]bool, len(values)) modelQuotaBlocks := make(map[uint64]account.ModelQuotaBlock, len(values)) @@ -459,6 +463,9 @@ func (r *AccountRepository) ListRoutingCandidates(ctx context.Context, provider if block, ok := modelQuotaBlocks[value.ID]; ok { candidate.ModelQuotaBlock = &block } + if block, ok := egressLeaseBlocks[value.ID]; ok { + candidate.EgressLeaseBlock = &block + } result = append(result, candidate) } return result, nil @@ -481,6 +488,10 @@ func (r *AccountRepository) ListRoutingAccountBases(ctx context.Context, provide if err != nil { return nil, err } + egressLeaseBlocks, err := r.getRoutingEgressLeaseBlocks(ctx, provider, values, time.Now().UTC()) + if err != nil { + return nil, err + } result := make([]account.RoutingAccountBase, 0, len(values)) for _, value := range values { base := account.RoutingAccountBase{Credential: value} @@ -493,11 +504,34 @@ func (r *AccountRepository) ListRoutingAccountBases(ctx context.Context, provide if window, ok := quotaWindows[value.ID]; ok { base.QuotaWindow = &window } + if block, ok := egressLeaseBlocks[value.ID]; ok { + base.EgressLeaseBlock = &block + } result = append(result, base) } return result, nil } +func (r *AccountRepository) getRoutingEgressLeaseBlocks(ctx context.Context, provider account.Provider, values []account.Credential, now time.Time) (map[uint64]account.EgressLeaseBlock, error) { + result := make(map[uint64]account.EgressLeaseBlock) + if len(values) == 0 { + return result, nil + } + var rows []accountEgressLeaseBlockModel + if err := r.db.db.WithContext(ctx). + Table("account_egress_lease_blocks AS block"). + Select("block.*"). + Joins("JOIN provider_accounts AS account ON account.id = block.account_id"). + Where("account.provider = ? AND account.enabled = ? AND account.auth_status = ? AND account.egress_node_id = block.node_id AND block.cooldown_until > ?", provider, true, account.AuthStatusActive, now.UTC()). + Find(&rows).Error; err != nil { + return nil, err + } + for _, row := range rows { + result[row.AccountID] = egressLeaseBlockFromModel(row) + } + return result, nil +} + // listRoutingCredentials loads only the account state required to decide which // account to use. Provider secrets deliberately stay in account_credentials // until a selected account is hydrated for the upstream call. @@ -1332,6 +1366,9 @@ func upsertKnownAccountByIdentity(tx *gorm.DB, value account.Credential, existin if err := tx.Save(&row).Error; err != nil { return repository.AccountUpsertResult{}, accountModel{}, err } + if _, err := deleteInvalidEgressLeaseBlocksForAccount(tx, row); err != nil { + return repository.AccountUpsertResult{}, accountModel{}, err + } if err := saveAccountRelations(tx, value, row.ID); err != nil { return repository.AccountUpsertResult{}, accountModel{}, err } @@ -1382,6 +1419,9 @@ func (r *AccountRepository) Update(ctx context.Context, value account.Credential if err := tx.Save(&row).Error; err != nil { return err } + if _, err := deleteInvalidEgressLeaseBlocksForAccount(tx, row); err != nil { + return err + } return saveAccountRelations(tx, value, row.ID) }); err != nil { return account.Credential{}, mapError(err) @@ -1577,6 +1617,11 @@ func (r *AccountRepository) UpdateMany(ctx context.Context, providerValue accoun } updated += result.RowsAffected } + if providerValue == account.ProviderBuild && updates.Enabled != nil && !*updates.Enabled { + if err := tx.Where("account_id IN ?", ids).Delete(&accountEgressLeaseBlockModel{}).Error; err != nil { + return err + } + } return nil }) if err != nil { @@ -1604,10 +1649,32 @@ func (r *AccountRepository) UpdateEgressBindings(ctx context.Context, providerVa values["egress_assignment_mode"] = string(mode) values["egress_assigned_at"] = assignedAt.UTC() } - result := r.db.db.WithContext(ctx).Model(&accountModel{}). - Where("provider = ? AND id IN ?", providerValue, ids). - Updates(values) - return result.RowsAffected, mapError(result.Error) + var updated int64 + var clearedLeaseBlocks int64 + err := r.db.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error { + result := tx.Model(&accountModel{}).Where("provider = ? AND id IN ?", providerValue, ids).Updates(values) + if result.Error != nil { + return result.Error + } + updated = result.RowsAffected + if providerValue != account.ProviderBuild { + return nil + } + query := tx.Where("account_id IN ?", ids) + if nodeID != nil { + query = query.Where("node_id <> ?", *nodeID) + } + deleted := query.Delete(&accountEgressLeaseBlockModel{}) + if deleted.Error != nil { + return deleted.Error + } + clearedLeaseBlocks = deleted.RowsAffected + return nil + }) + if err == nil && clearedLeaseBlocks > 0 { + r.notifyInvalidation(ctx, repository.InvalidationEvent{Kind: repository.InvalidationAccountEgressLeaseChanged, Provider: providerValue}) + } + return updated, mapError(err) } // ListEgressAssignments returns all accounts for one provider with their @@ -2163,6 +2230,181 @@ func (r *AccountRepository) UpsertModelQuotaBlock(ctx context.Context, value acc return err } +func egressLeaseBlockFromModel(row accountEgressLeaseBlockModel) account.EgressLeaseBlock { + return account.EgressLeaseBlock{ + AccountID: row.AccountID, NodeID: row.NodeID, Reason: row.Reason, Version: row.Version, + CooldownUntil: row.CooldownUntil.UTC(), UpdatedAt: row.UpdatedAt.UTC(), + } +} + +// ListEgressLeaseBlocks returns the durable guard-owned lease state, including +// expired rows. The sidecar uses expired rows for recovery reconciliation; the +// selector independently ignores them after CooldownUntil as a fail-safe. +func (r *AccountRepository) ListEgressLeaseBlocks(ctx context.Context, limit int, after *account.EgressLeaseBlockCursor) ([]account.EgressLeaseBlock, error) { + if limit <= 0 || limit > 1001 { + return nil, repository.ErrConflict + } + var rows []accountEgressLeaseBlockModel + query := r.db.db.WithContext(ctx). + Table("account_egress_lease_blocks AS block").Select("block.*"). + Joins("JOIN provider_accounts AS account ON account.id = block.account_id"). + Joins("JOIN egress_nodes AS node ON node.id = block.node_id"). + Where("account.provider = ? AND account.enabled = ? AND account.auth_status = ? AND account.egress_node_id = block.node_id AND node.enabled = ? AND node.scope = ?", account.ProviderBuild, true, account.AuthStatusActive, true, "grok_build"). + Order("block.cooldown_until ASC, block.account_id ASC, block.node_id ASC").Limit(limit) + if after != nil { + cursorTime := after.CooldownUntil.UTC() + query = query.Where( + "block.cooldown_until > ? OR (block.cooldown_until = ? AND (block.account_id > ? OR (block.account_id = ? AND block.node_id > ?)))", + cursorTime, cursorTime, after.AccountID, after.AccountID, after.NodeID, + ) + } + if err := query.Find(&rows).Error; err != nil { + return nil, err + } + values := make([]account.EgressLeaseBlock, 0, len(rows)) + for _, row := range rows { + values = append(values, egressLeaseBlockFromModel(row)) + } + return values, nil +} + +func deleteInvalidEgressLeaseBlocksForAccount(tx *gorm.DB, row accountModel) (int64, error) { + if account.Provider(row.Provider) != account.ProviderBuild { + return 0, nil + } + query := tx.Where("account_id = ?", row.ID) + if row.Enabled && account.AuthStatus(row.AuthStatus) == account.AuthStatusActive && row.EgressNodeID != nil { + query = query.Where("node_id <> ?", *row.EgressNodeID) + } + result := query.Delete(&accountEgressLeaseBlockModel{}) + return result.RowsAffected, result.Error +} + +func (r *AccountRepository) PruneInvalidEgressLeaseBlocks(ctx context.Context, limit int) (int64, error) { + if limit < 1 || limit > 1000 { + return 0, repository.ErrConflict + } + var rows []accountEgressLeaseBlockModel + err := r.db.db.WithContext(ctx). + Table("account_egress_lease_blocks AS block").Select("block.*"). + Joins("LEFT JOIN provider_accounts AS account ON account.id = block.account_id"). + Joins("LEFT JOIN egress_nodes AS node ON node.id = block.node_id"). + Where("account.id IS NULL OR account.provider <> ? OR account.enabled <> ? OR account.auth_status <> ? OR account.egress_node_id IS NULL OR account.egress_node_id <> block.node_id OR node.id IS NULL OR node.enabled <> ? OR node.scope <> ?", account.ProviderBuild, true, account.AuthStatusActive, true, "grok_build"). + Order("block.cooldown_until ASC, block.account_id ASC, block.node_id ASC").Limit(limit).Find(&rows).Error + if err != nil || len(rows) == 0 { + return 0, err + } + var deleted int64 + err = r.db.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error { + for start := 0; start < len(rows); start += 400 { + end := min(start+400, len(rows)) + pairs := make([][]any, 0, end-start) + for _, row := range rows[start:end] { + pairs = append(pairs, []any{row.AccountID, row.NodeID}) + } + result := tx.Where("(account_id, node_id) IN ?", pairs).Delete(&accountEgressLeaseBlockModel{}) + if result.Error != nil { + return result.Error + } + deleted += result.RowsAffected + } + return nil + }) + if err == nil && deleted > 0 { + r.notifyInvalidation(ctx, repository.InvalidationEvent{Kind: repository.InvalidationAccountEgressLeaseChanged, Provider: account.ProviderBuild}) + } + return deleted, err +} + +func (r *AccountRepository) DeleteEgressLeaseBlocksByNodes(ctx context.Context, nodeIDs []uint64) (int64, error) { + if len(nodeIDs) == 0 { + return 0, nil + } + result := r.db.db.WithContext(ctx).Where("node_id IN ?", nodeIDs).Delete(&accountEgressLeaseBlockModel{}) + if result.Error == nil && result.RowsAffected > 0 { + r.notifyInvalidation(ctx, repository.InvalidationEvent{Kind: repository.InvalidationAccountEgressLeaseChanged, Provider: account.ProviderBuild}) + } + return result.RowsAffected, result.Error +} + +// UpsertEgressLeaseBlock atomically verifies that the Build account is still +// bound to the requested node. A shorter concurrent hold cannot replace a +// longer one or rotate its CAS version. +func (r *AccountRepository) UpsertEgressLeaseBlock(ctx context.Context, value account.EgressLeaseBlock) (account.EgressLeaseBlock, error) { + value.Reason = strings.TrimSpace(value.Reason) + value.Version = strings.TrimSpace(value.Version) + if value.AccountID == 0 || value.NodeID == 0 || value.Reason == "" || len(value.Version) < 16 || len(value.Version) > 64 || value.CooldownUntil.IsZero() { + return account.EgressLeaseBlock{}, repository.ErrConflict + } + value.Reason = truncate(value.Reason, 100) + value.CooldownUntil = value.CooldownUntil.UTC() + value.UpdatedAt = time.Now().UTC() + stored := value + err := r.db.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error { + var owner accountModel + if err := tx.Clauses(clause.Locking{Strength: "UPDATE"}).Select("id", "provider", "enabled", "auth_status", "egress_node_id").First(&owner, value.AccountID).Error; err != nil { + return mapError(err) + } + if account.Provider(owner.Provider) != account.ProviderBuild || !owner.Enabled || account.AuthStatus(owner.AuthStatus) != account.AuthStatusActive || owner.EgressNodeID == nil || *owner.EgressNodeID != value.NodeID { + return repository.ErrConflict + } + var existing accountEgressLeaseBlockModel + load := tx.Where("account_id = ? AND node_id = ?", value.AccountID, value.NodeID).Limit(1).Find(&existing) + if load.Error != nil { + return load.Error + } + if load.RowsAffected > 0 && existing.CooldownUntil.After(value.CooldownUntil) { + stored = egressLeaseBlockFromModel(existing) + return nil + } + row := accountEgressLeaseBlockModel{ + AccountID: value.AccountID, NodeID: value.NodeID, Reason: value.Reason, Version: value.Version, + CooldownUntil: value.CooldownUntil, UpdatedAt: value.UpdatedAt, + } + created := tx.Clauses(clause.OnConflict{ + Columns: []clause.Column{{Name: "account_id"}, {Name: "node_id"}}, + DoUpdates: clause.AssignmentColumns([]string{"reason", "version", "cooldown_until", "updated_at"}), + Where: clause.Where{Exprs: []clause.Expression{clause.Expr{ + SQL: "account_egress_lease_blocks.cooldown_until <= excluded.cooldown_until", + }}}, + }).Create(&row) + if created.Error != nil { + return created.Error + } + if created.RowsAffected == 0 { + if err := tx.Where("account_id = ? AND node_id = ?", value.AccountID, value.NodeID).First(&existing).Error; err != nil { + return err + } + stored = egressLeaseBlockFromModel(existing) + return nil + } + stored = egressLeaseBlockFromModel(row) + return nil + }) + if err == nil { + r.notifyInvalidation(ctx, repository.InvalidationEvent{Kind: repository.InvalidationAccountEgressLeaseChanged, Provider: account.ProviderBuild, AccountID: value.AccountID}) + } + return stored, err +} + +// DeleteEgressLeaseBlock uses the opaque version as a compare-and-swap token, +// so a stale recovery probe cannot clear a newer quarantine. +func (r *AccountRepository) DeleteEgressLeaseBlock(ctx context.Context, accountID, nodeID uint64, version string) (bool, error) { + version = strings.TrimSpace(version) + if accountID == 0 || nodeID == 0 || version == "" { + return false, repository.ErrConflict + } + result := r.db.db.WithContext(ctx).Where("account_id = ? AND node_id = ? AND version = ?", accountID, nodeID, version).Delete(&accountEgressLeaseBlockModel{}) + if result.Error != nil { + return false, result.Error + } + if result.RowsAffected == 1 { + r.notifyInvalidation(ctx, repository.InvalidationEvent{Kind: repository.InvalidationAccountEgressLeaseChanged, Provider: account.ProviderBuild, AccountID: accountID}) + return true, nil + } + return false, nil +} + func (r *AccountRepository) PruneExpiredModelQuotaBlocks(ctx context.Context, now time.Time, limit int) (int64, error) { if limit <= 0 || limit > 1000 { limit = 100 diff --git a/backend/internal/infra/persistence/relational/models.go b/backend/internal/infra/persistence/relational/models.go index 4de60c6d1..48b3fec38 100644 --- a/backend/internal/infra/persistence/relational/models.go +++ b/backend/internal/infra/persistence/relational/models.go @@ -243,6 +243,19 @@ type accountModelQuotaBlockModel struct { func (accountModelQuotaBlockModel) TableName() string { return "account_model_quota_blocks" } +type accountEgressLeaseBlockModel struct { + AccountID uint64 `gorm:"primaryKey"` + NodeID uint64 `gorm:"primaryKey"` + Reason string `gorm:"size:100;not null;check:chk_account_egress_lease_blocks_reason,length(trim(reason)) BETWEEN 1 AND 100"` + Version string `gorm:"size:64;not null;check:chk_account_egress_lease_blocks_version,length(trim(version)) BETWEEN 16 AND 64"` + CooldownUntil time.Time `gorm:"not null"` + UpdatedAt time.Time `gorm:"not null"` + Account *accountModel `gorm:"foreignKey:AccountID;references:ID;constraint:OnUpdate:CASCADE,OnDelete:CASCADE"` + Node *egressNodeModel `gorm:"foreignKey:NodeID;references:ID;constraint:OnUpdate:CASCADE,OnDelete:CASCADE"` +} + +func (accountEgressLeaseBlockModel) TableName() string { return "account_egress_lease_blocks" } + type clientKeyModel struct { ID uint64 `gorm:"primaryKey;autoIncrement"` Name string `gorm:"size:160;not null;check:chk_client_keys_name,length(trim(name)) BETWEEN 1 AND 160"` diff --git a/backend/internal/infra/persistence/relational/schema.go b/backend/internal/infra/persistence/relational/schema.go index 8ef00f589..6ac135382 100644 --- a/backend/internal/infra/persistence/relational/schema.go +++ b/backend/internal/infra/persistence/relational/schema.go @@ -42,6 +42,7 @@ var schemaModels = []any{ &accountModelCapabilityModel{}, &accountModelSyncStateModel{}, &accountModelQuotaBlockModel{}, + &accountEgressLeaseBlockModel{}, &clientKeyModel{}, &clientKeyModelPermission{}, &billingReservationModel{}, @@ -67,6 +68,7 @@ var schemaIndexes = []string{ "CREATE INDEX IF NOT EXISTS idx_accounts_auto_clean_reauth_cursor ON provider_accounts(auth_status, enabled, id, reauth_marked_at)", "CREATE INDEX IF NOT EXISTS idx_account_credentials_refresh_due ON account_credentials(refresh_due_at, account_id)", "CREATE INDEX IF NOT EXISTS idx_account_credentials_build_bot_flag ON account_credentials(build_bot_flag_source, account_id)", + "CREATE INDEX IF NOT EXISTS idx_account_egress_lease_blocks_due ON account_egress_lease_blocks(cooldown_until, account_id, node_id)", "CREATE INDEX IF NOT EXISTS idx_quota_windows_due ON account_quota_windows(remaining, reset_at, account_id)", "CREATE INDEX IF NOT EXISTS idx_model_routes_public_id_lookup ON model_routes(public_id)", // Catalog/discovered rows remain idempotent per API capability. One public diff --git a/backend/internal/repository/runtime.go b/backend/internal/repository/runtime.go index a5dd5d349..ec67889bd 100644 --- a/backend/internal/repository/runtime.go +++ b/backend/internal/repository/runtime.go @@ -96,14 +96,15 @@ const ( // InvalidationAccountHealthChanged carries the exact request-path health // mutation for one account. Unlike an arbitrary account state change, it can // be applied as a small runtime overlay without rebuilding the provider pool. - InvalidationAccountHealthChanged InvalidationKind = "account_health_changed" - InvalidationAccountCredentialChanged InvalidationKind = "account_credential_changed" - InvalidationAccountCapabilityChanged InvalidationKind = "account_capability_changed" - InvalidationAccountBillingChanged InvalidationKind = "account_billing_changed" - InvalidationAccountQuotaChanged InvalidationKind = "account_quota_changed" - InvalidationAccountRecoveryChanged InvalidationKind = "account_recovery_changed" - InvalidationAccountModelQuotaChanged InvalidationKind = "account_model_quota_changed" - InvalidationClientKeyChanged InvalidationKind = "client_key_changed" + InvalidationAccountHealthChanged InvalidationKind = "account_health_changed" + InvalidationAccountCredentialChanged InvalidationKind = "account_credential_changed" + InvalidationAccountCapabilityChanged InvalidationKind = "account_capability_changed" + InvalidationAccountBillingChanged InvalidationKind = "account_billing_changed" + InvalidationAccountQuotaChanged InvalidationKind = "account_quota_changed" + InvalidationAccountRecoveryChanged InvalidationKind = "account_recovery_changed" + InvalidationAccountEgressLeaseChanged InvalidationKind = "account_egress_lease_changed" + InvalidationAccountModelQuotaChanged InvalidationKind = "account_model_quota_changed" + InvalidationClientKeyChanged InvalidationKind = "client_key_changed" ) type InvalidationLayer string @@ -137,7 +138,7 @@ func (e InvalidationEvent) Layer() InvalidationLayer { return InvalidationLayerRoute case InvalidationModelBindingChanged, InvalidationAccountCapabilityChanged, InvalidationAccountModelQuotaChanged: return InvalidationLayerOverlay - case InvalidationAccountStateChanged, InvalidationAccountHealthChanged, InvalidationAccountCredentialChanged, InvalidationAccountBillingChanged, InvalidationAccountQuotaChanged, InvalidationAccountRecoveryChanged: + case InvalidationAccountStateChanged, InvalidationAccountHealthChanged, InvalidationAccountCredentialChanged, InvalidationAccountBillingChanged, InvalidationAccountQuotaChanged, InvalidationAccountRecoveryChanged, InvalidationAccountEgressLeaseChanged: return InvalidationLayerBase case InvalidationClientKeyChanged: return InvalidationLayerClientKey diff --git a/backend/internal/transport/http/audit/handler.go b/backend/internal/transport/http/audit/handler.go index 88ce7a750..4cb5d4bf2 100644 --- a/backend/internal/transport/http/audit/handler.go +++ b/backend/internal/transport/http/audit/handler.go @@ -43,6 +43,7 @@ type qualityGuardAuditResponse struct { RequestID string `json:"requestId"` QualityProbe bool `json:"qualityProbe"` Provider string `json:"provider"` + AccountID *uint64 `json:"accountId,string,omitempty"` EgressNodeID *uint64 `json:"egressNodeId,string,omitempty"` EgressNodeName string `json:"egressNodeName,omitempty"` StatusCode int `json:"statusCode"` @@ -74,7 +75,7 @@ func (h *Handler) listQualityGuard(c *gin.Context) { for _, value := range result.Items { items = append(items, qualityGuardAuditResponse{ ID: value.ID, RequestID: value.RequestID, QualityProbe: value.ClientKeyID == h.qualityGuardClientKeyID, - Provider: value.Provider, EgressNodeID: value.EgressNodeID, EgressNodeName: value.EgressNodeName, + Provider: value.Provider, AccountID: value.AccountID, EgressNodeID: value.EgressNodeID, EgressNodeName: value.EgressNodeName, StatusCode: value.StatusCode, Streaming: value.Streaming, OutputTokens: value.OutputTokens, ReasoningTokens: value.ReasoningTokens, FirstTokenMS: value.FirstTokenMS, DurationMS: value.DurationMS, ErrorCode: value.ErrorCode, diff --git a/backend/internal/transport/http/egress/handler.go b/backend/internal/transport/http/egress/handler.go index 5b602eb38..631456860 100644 --- a/backend/internal/transport/http/egress/handler.go +++ b/backend/internal/transport/http/egress/handler.go @@ -1,6 +1,7 @@ package egress import ( + "encoding/base64" "encoding/json" "errors" "fmt" @@ -97,6 +98,9 @@ func (h *Handler) RegisterQualityGuard(router *gin.RouterGroup) { router.PATCH("/egress-nodes/batch", h.updateMany) router.POST("/egress-nodes/:id/test", h.testNode) router.POST("/egress-nodes/:id/quality-test", h.testQualityGuardNode) + router.GET("/egress-leases", h.listQualityGuardLeases) + router.POST("/egress-leases/quarantine", h.quarantineQualityGuardLease) + router.POST("/egress-leases/restore", h.restoreQualityGuardLease) router.GET("/egress-operations", h.operationsConfig) } @@ -159,6 +163,9 @@ type qualityGuardConfig struct { } type qualityGuardNodeState struct { + ObserveOnly bool `json:"observe_only"` + ObserveOnlyReason string `json:"observe_only_reason"` + QuarantinedLeases int `json:"quarantined_lease_count"` ActiveSoftStrikes int `json:"active_soft_strikes"` PassiveSoftStrikes int `json:"passive_soft_strikes"` ErrorStrikes int `json:"error_strikes"` @@ -180,9 +187,12 @@ type qualityGuardEvent struct { Event string `json:"event"` NodeID string `json:"node_id"` NodeName string `json:"node_name"` + AccountID string `json:"account_id,omitempty"` + RequestID string `json:"request_id,omitempty"` Reason string `json:"reason"` Classification string `json:"classification"` OutputTPS float64 `json:"output_tps"` + CooldownUntil float64 `json:"cooldown_until,omitempty"` } func (h *Handler) qualityGuardStatus(c *gin.Context) { @@ -384,6 +394,7 @@ func (h *Handler) testQualityGuardNode(c *gin.Context) { } var request struct { ProfileID string `json:"profileId"` + AccountID string `json:"accountId"` } _ = c.ShouldBindJSON(&request) input, err := h.resolveProbeInput(strings.TrimSpace(request.ProfileID)) @@ -395,6 +406,14 @@ func (h *Handler) testQualityGuardNode(c *gin.Context) { response.Error(c, http.StatusServiceUnavailable, "qualityGuardUnavailable", "质量守护配置暂不可用") return } + if strings.TrimSpace(request.AccountID) != "" { + accountID, parseErr := strconv.ParseUint(request.AccountID, 10, 64) + if parseErr != nil || accountID == 0 { + response.Error(c, http.StatusBadRequest, "invalidAccountId", "账号 ID 无效") + return + } + input.AccountID = accountID + } value, err := h.service.ProbeQuality(c.Request.Context(), nodeID, input) if err != nil { h.writeQualityProbeError(c, err) @@ -413,6 +432,149 @@ func (h *Handler) testQualityGuardNode(c *gin.Context) { }) } +type qualityLeaseRequest struct { + AccountID string `json:"accountId" binding:"required"` + NodeID string `json:"nodeId" binding:"required"` + Reason string `json:"reason"` + Version string `json:"version"` + QuarantineSeconds int `json:"quarantineSeconds"` +} + +type qualityLeaseCursor struct { + CooldownUntil int64 `json:"t"` + AccountID uint64 `json:"a"` + NodeID uint64 `json:"n"` +} + +func qualityLeaseResponse(value accountdomain.EgressLeaseBlock) gin.H { + return gin.H{ + "accountId": strconv.FormatUint(value.AccountID, 10), "nodeId": strconv.FormatUint(value.NodeID, 10), + "reason": value.Reason, "version": value.Version, "cooldownUntil": float64(value.CooldownUntil.UnixMilli()) / 1000, + "updatedAt": value.UpdatedAt.UTC(), + } +} + +func (h *Handler) listQualityGuardLeases(c *gin.Context) { + limit, parseErr := strconv.Atoi(c.DefaultQuery("limit", "500")) + if parseErr != nil || limit < 1 || limit > 1000 { + response.Error(c, http.StatusBadRequest, "invalidPageSize", "分页大小无效") + return + } + cursor, cursorErr := decodeQualityLeaseCursor(c.Query("cursor")) + if cursorErr != nil { + response.Error(c, http.StatusBadRequest, "invalidCursor", "分页游标无效") + return + } + values, err := h.service.ListQualityLeases(c.Request.Context(), limit+1, cursor) + if err != nil { + h.writeQualityLeaseError(c, err) + return + } + hasMore := len(values) > limit + if hasMore { + values = values[:limit] + } + items := make([]gin.H, 0, len(values)) + for _, value := range values { + items = append(items, qualityLeaseResponse(value)) + } + nextCursor := "" + if hasMore && len(values) > 0 { + nextCursor = encodeQualityLeaseCursor(values[len(values)-1]) + } + response.Success(c, http.StatusOK, gin.H{"items": items, "hasMore": hasMore, "nextCursor": nextCursor}) +} + +func decodeQualityLeaseCursor(raw string) (*accountdomain.EgressLeaseBlockCursor, error) { + raw = strings.TrimSpace(raw) + if raw == "" { + return nil, nil + } + if len(raw) > 256 { + return nil, errors.New("cursor too long") + } + decoded, err := base64.RawURLEncoding.DecodeString(raw) + if err != nil { + return nil, err + } + var value qualityLeaseCursor + decoder := json.NewDecoder(strings.NewReader(string(decoded))) + decoder.DisallowUnknownFields() + if err := decoder.Decode(&value); err != nil || value.CooldownUntil <= 0 || value.AccountID == 0 || value.NodeID == 0 { + return nil, errors.New("invalid cursor") + } + if err := decoder.Decode(&struct{}{}); !errors.Is(err, io.EOF) { + return nil, errors.New("invalid cursor") + } + return &accountdomain.EgressLeaseBlockCursor{ + CooldownUntil: time.Unix(0, value.CooldownUntil).UTC(), AccountID: value.AccountID, NodeID: value.NodeID, + }, nil +} + +func encodeQualityLeaseCursor(value accountdomain.EgressLeaseBlock) string { + payload, _ := json.Marshal(qualityLeaseCursor{ + CooldownUntil: value.CooldownUntil.UTC().UnixNano(), AccountID: value.AccountID, NodeID: value.NodeID, + }) + return base64.RawURLEncoding.EncodeToString(payload) +} + +func (h *Handler) quarantineQualityGuardLease(c *gin.Context) { + var request qualityLeaseRequest + if c.ShouldBindJSON(&request) != nil { + response.Error(c, http.StatusBadRequest, "invalidRequest", "请求参数无效") + return + } + accountID, accountErr := strconv.ParseUint(request.AccountID, 10, 64) + nodeID, nodeErr := strconv.ParseUint(request.NodeID, 10, 64) + if accountErr != nil || nodeErr != nil || accountID == 0 || nodeID == 0 { + response.Error(c, http.StatusBadRequest, "invalidId", "账号或节点 ID 无效") + return + } + value, err := h.service.QuarantineQualityLease(c.Request.Context(), egressapp.QualityLeaseInput{ + AccountID: accountID, NodeID: nodeID, Reason: request.Reason, QuarantineSeconds: request.QuarantineSeconds, + }) + if err != nil { + h.writeQualityLeaseError(c, err) + return + } + response.Success(c, http.StatusOK, qualityLeaseResponse(value)) +} + +func (h *Handler) restoreQualityGuardLease(c *gin.Context) { + var request qualityLeaseRequest + if c.ShouldBindJSON(&request) != nil { + response.Error(c, http.StatusBadRequest, "invalidRequest", "请求参数无效") + return + } + accountID, accountErr := strconv.ParseUint(request.AccountID, 10, 64) + nodeID, nodeErr := strconv.ParseUint(request.NodeID, 10, 64) + if accountErr != nil || nodeErr != nil || accountID == 0 || nodeID == 0 { + response.Error(c, http.StatusBadRequest, "invalidId", "账号或节点 ID 无效") + return + } + restored, err := h.service.RestoreQualityLease(c.Request.Context(), accountID, nodeID, request.Version) + if err != nil { + h.writeQualityLeaseError(c, err) + return + } + response.Success(c, http.StatusOK, gin.H{"restored": restored}) +} + +func (h *Handler) writeQualityLeaseError(c *gin.Context, err error) { + switch { + case errors.Is(err, egressapp.ErrInvalidInput): + response.Error(c, http.StatusBadRequest, "invalidQualityLease", "租约隔离参数无效") + case errors.Is(err, egressapp.ErrNotFound): + response.Error(c, http.StatusNotFound, "egressNodeNotFound", "代理节点不存在") + case errors.Is(err, egressapp.ErrQualityLeaseConflict): + response.Error(c, http.StatusConflict, "qualityLeaseConflict", "租约绑定或隔离版本已变化") + case errors.Is(err, egressapp.ErrQualityLeaseUnavailable): + response.Error(c, http.StatusServiceUnavailable, "qualityLeaseUnavailable", "租约级质量隔离暂不可用") + default: + response.Error(c, http.StatusInternalServerError, "qualityLeaseOperationFailed", "租约级质量隔离操作失败") + } +} + func (h *Handler) cleanupPreview(c *gin.Context) { value, err := h.service.PreviewUnhealthyCleanup(c.Request.Context()) if err != nil { diff --git a/backend/internal/transport/http/egress/handler_test.go b/backend/internal/transport/http/egress/handler_test.go index cec713d04..dfeb1599c 100644 --- a/backend/internal/transport/http/egress/handler_test.go +++ b/backend/internal/transport/http/egress/handler_test.go @@ -14,12 +14,38 @@ import ( "time" egressapp "github.com/chenyme/grok2api/backend/internal/application/egress" + accountdomain "github.com/chenyme/grok2api/backend/internal/domain/account" egressdomain "github.com/chenyme/grok2api/backend/internal/domain/egress" "github.com/chenyme/grok2api/backend/internal/infra/security" "github.com/chenyme/grok2api/backend/internal/repository" "github.com/gin-gonic/gin" ) +func TestQualityLeaseCursorRoundTripAndValidation(t *testing.T) { + want := accountdomain.EgressLeaseBlock{ + AccountID: 42, + NodeID: 9, + CooldownUntil: time.Date(2026, time.August, 19, 8, 7, 6, 123456789, time.UTC), + } + encoded := encodeQualityLeaseCursor(want) + got, err := decodeQualityLeaseCursor(encoded) + if err != nil { + t.Fatal(err) + } + if got.AccountID != want.AccountID || got.NodeID != want.NodeID || !got.CooldownUntil.Equal(want.CooldownUntil) { + t.Fatalf("cursor = %#v, want %#v", got, want) + } + for _, raw := range []string{ + "not-base64", + "eyJ0IjoxLCJhIjowLCJuIjoxfQ", + "eyJ0IjoxLCJhIjoxLCJuIjoxfXsic2Vjb25kIjp0cnVlfQ", + } { + if _, err := decodeQualityLeaseCursor(raw); err == nil { + t.Fatalf("decodeQualityLeaseCursor(%q) succeeded", raw) + } + } +} + type proxyRevealRepository struct { node egressdomain.Node profile egressdomain.ProxyProfile @@ -131,7 +157,7 @@ func TestProxyProfileListUsesBoundedPaginationAndSearch(t *testing.T) { func TestQualityGuardStatusReadsOnlyPublicState(t *testing.T) { path := t.TempDir() + "/state.json" - state := `{"version":1,"started_at":10,"updated_at":20,"last_active_cycle_at":15,"last_passive_poll_at":19,"password":"must-not-leak","guard":{"mode":"hybrid","model":"grok-4.5","client_key_id":"6","node_ids":["8"],"active_interval_seconds":1800,"passive_poll_seconds":5,"soft_tps":500,"hard_tps":1000,"consecutive_soft":2,"consecutive_errors":2,"quarantine_seconds":300,"min_healthy_nodes":3,"max_output_tokens":384,"fail_closed":true,"min_generation_ms":1234,"prompt":"private-probe-prompt","expected":"private-marker"},"protected_node_ids":["9"],"nodes":{"8":{"active_soft_strikes":0,"passive_soft_strikes":0,"error_strikes":0,"quarantined_until":0,"disabled_by_guard":false,"last_reason":"","last_probe_at":15,"last_observed_at":19,"last_source":"passive","last_classification":"healthy","last_output_tps":42.5,"last_output_tokens":100,"last_first_token_ms":900,"last_duration_ms":4000}},"statistics":{"started_at":11,"active":{"total":7,"healthy":6,"soft":1,"hard":0,"errors":0,"output_tokens":1400},"passive":{"total":9,"healthy":8,"soft":0,"hard":1,"errors":0,"output_tokens":1800},"actions":{"quarantined":1,"restored":0,"suppressed":0}}}` + state := `{"version":1,"started_at":10,"updated_at":20,"last_active_cycle_at":15,"last_passive_poll_at":19,"password":"must-not-leak","guard":{"mode":"hybrid","model":"grok-4.5","client_key_id":"6","node_ids":["8"],"active_interval_seconds":1800,"passive_poll_seconds":5,"soft_tps":500,"hard_tps":1000,"consecutive_soft":2,"consecutive_errors":2,"quarantine_seconds":300,"min_healthy_nodes":3,"max_output_tokens":384,"fail_closed":true,"min_generation_ms":1234,"prompt":"private-probe-prompt","expected":"private-marker"},"protected_node_ids":["9"],"nodes":{"8":{"observe_only":true,"observe_only_reason":"account_bound_proxy","active_soft_strikes":0,"passive_soft_strikes":0,"error_strikes":0,"quarantined_until":0,"disabled_by_guard":false,"last_reason":"","last_probe_at":15,"last_observed_at":19,"last_source":"passive","last_classification":"healthy","last_output_tps":42.5,"last_output_tokens":100,"last_first_token_ms":900,"last_duration_ms":4000}},"statistics":{"started_at":11,"active":{"total":7,"healthy":6,"soft":1,"hard":0,"errors":0,"output_tokens":1400},"passive":{"total":9,"healthy":8,"soft":0,"hard":1,"errors":0,"output_tokens":1800},"actions":{"quarantined":1,"restored":0,"suppressed":0}}}` if err := os.WriteFile(path, []byte(state), 0o600); err != nil { t.Fatal(err) } @@ -139,7 +165,7 @@ func TestQualityGuardStatusReadsOnlyPublicState(t *testing.T) { context, _ := gin.CreateTestContext(recorder) context.Request = httptest.NewRequest("GET", "/egress-quality-guard", nil) NewHandler(nil, path).qualityGuardStatus(context) - if recorder.Code != 200 || !strings.Contains(recorder.Body.String(), `"available":true`) || !strings.Contains(recorder.Body.String(), `"last_output_tps":42.5`) || !strings.Contains(recorder.Body.String(), `"output_tokens":1400`) || !strings.Contains(recorder.Body.String(), `"protectedNodeIds":["9"]`) || !strings.Contains(recorder.Body.String(), `"fail_closed":true`) || !strings.Contains(recorder.Body.String(), `"min_generation_ms":1234`) { + if recorder.Code != 200 || !strings.Contains(recorder.Body.String(), `"available":true`) || !strings.Contains(recorder.Body.String(), `"observe_only":true`) || !strings.Contains(recorder.Body.String(), `"observe_only_reason":"account_bound_proxy"`) || !strings.Contains(recorder.Body.String(), `"last_output_tps":42.5`) || !strings.Contains(recorder.Body.String(), `"output_tokens":1400`) || !strings.Contains(recorder.Body.String(), `"protectedNodeIds":["9"]`) || !strings.Contains(recorder.Body.String(), `"fail_closed":true`) || !strings.Contains(recorder.Body.String(), `"min_generation_ms":1234`) { t.Fatalf("status=%d body=%s", recorder.Code, recorder.Body.String()) } if strings.Contains(recorder.Body.String(), "must-not-leak") || strings.Contains(recorder.Body.String(), "private-probe-prompt") || strings.Contains(recorder.Body.String(), "private-marker") || strings.Contains(recorder.Body.String(), "client_key_id") || !strings.Contains(recorder.Body.String(), `"recentEvents":[]`) { diff --git a/frontend/src/features/quality-guard/quality-guard-api.ts b/frontend/src/features/quality-guard/quality-guard-api.ts index 39a49b4bd..3c7aa9dba 100644 --- a/frontend/src/features/quality-guard/quality-guard-api.ts +++ b/frontend/src/features/quality-guard/quality-guard-api.ts @@ -14,6 +14,9 @@ export type QualityGuardPolicy = { }; export type QualityGuardNodeState = { + observe_only?: boolean; + observe_only_reason?: string; + quarantined_lease_count?: number; active_soft_strikes: number; passive_soft_strikes: number; error_strikes: number; @@ -35,9 +38,12 @@ export type QualityGuardEvent = { event: string; node_id: string; node_name: string; + account_id?: string; + request_id?: string; reason: string; classification: string; output_tps: number; + cooldown_until?: number; }; export type QualityGuardDetectionStats = { @@ -122,6 +128,7 @@ export type QualityTestResult = { }; const nodeStateValidator = hasShape({ + observe_only: isOptional(isBoolean), observe_only_reason: isOptional(isString), quarantined_lease_count: isOptional(isNumber), active_soft_strikes: isNumber, passive_soft_strikes: isNumber, error_strikes: isNumber, quarantined_until: isNumber, disabled_by_guard: isBoolean, last_reason: isString, last_probe_at: isNumber, last_observed_at: isNumber, last_source: isString, @@ -130,7 +137,9 @@ const nodeStateValidator = hasShape({ }); const eventValidator = hasShape({ ts: isNumber, event: isString, node_id: isString, node_name: isString, + account_id: isOptional(isString), request_id: isOptional(isString), reason: isString, classification: isString, output_tps: isNumber, + cooldown_until: isOptional(isNumber), }); const configValidator = hasShape({ mode: isOneOf("active", "passive", "hybrid"), model: isString, diff --git a/frontend/src/features/quality-guard/quality-guard-page.tsx b/frontend/src/features/quality-guard/quality-guard-page.tsx index a754e13f6..84d610531 100644 --- a/frontend/src/features/quality-guard/quality-guard-page.tsx +++ b/frontend/src/features/quality-guard/quality-guard-page.tsx @@ -147,6 +147,7 @@ export function QualityGuardPage() { const fresh = isFresh(status); const guardedNodes = status?.nodes ?? {}; const quarantined = Object.values(guardedNodes).filter((node) => node.disabled_by_guard).length; + const quarantinedLeases = Object.values(guardedNodes).reduce((total, node) => total + (node.quarantined_lease_count ?? 0), 0); const enabled = nodes.filter((node) => node.enabled).length; return ( @@ -186,7 +187,7 @@ export function QualityGuardPage() { - + {status.statistics ? : null} @@ -261,7 +262,7 @@ function StatisticsPanel({ statistics, locale }: { statistics: QualityGuardStati { icon: Eye, label: t("qualityGuard.statisticsPassive"), value: formatCount(statistics.passive.total, locale), detail: t("qualityGuard.statisticsPassiveDetail", { healthy: formatCount(statistics.passive.healthy, locale) }) }, { icon: Coins, label: t("qualityGuard.statisticsTokens"), value: formatCount(statistics.active.output_tokens, locale), detail: t("qualityGuard.statisticsTokensHelp") }, { icon: AlertTriangle, label: t("qualityGuard.statisticsAnomalies"), value: formatCount(anomalies, locale), detail: t("qualityGuard.statisticsAnomalyDetail", { soft: formatCount(statistics.active.soft + statistics.passive.soft, locale), hard: formatCount(statistics.active.hard + statistics.passive.hard, locale) }) }, - { icon: Shield, label: t("qualityGuard.statisticsQuarantines"), value: formatCount(statistics.actions.quarantined, locale), detail: t("qualityGuard.statisticsActionDetail", { restored: formatCount(statistics.actions.restored, locale), suppressed: formatCount(statistics.actions.suppressed, locale) }) }, + { icon: Shield, label: t("qualityGuard.statisticsQuarantines"), value: formatCount(statistics.actions.quarantined, locale), detail: t("qualityGuard.statisticsSuppressedActionDetail", { restored: formatCount(statistics.actions.restored, locale), suppressed: formatCount(statistics.actions.suppressed, locale) }) }, ]; return
@@ -377,6 +378,9 @@ function StateBadge({ node, state, protectedNode }: { node: EgressNodeDTO; state if (state?.disabled_by_guard) return {t("qualityGuard.quarantined")}; if (protectedNode) return {t("qualityGuard.fixedFallback")}; if (!node.enabled) return {t("common.disabled")}; + if (state?.quarantined_lease_count) return {t("qualityGuard.leaseQuarantined", { count: state.quarantined_lease_count })}; + if (state?.observe_only) return {t("qualityGuard.leaseScopedObserveOnly")}; + if (node.accountBoundProxy) return {t("qualityGuard.leaseScoped")}; if (state?.error_strikes) return {t("qualityGuard.probeFailed")}; if (state?.last_classification === "hard" || state?.last_classification === "soft") return {t("qualityGuard.suspect")}; if (state?.last_classification === "healthy") return {t("qualityGuard.healthy")}; @@ -389,13 +393,29 @@ function EventList({ events, locale }: { events: QualityGuardEvent[]; locale: st

{t("qualityGuard.events")}

{events.length === 0 ?

{t("qualityGuard.noEvents")}

:
{[...events].reverse().slice(0, 10).map((event, index) =>
-

{event.node_name || `ID ${event.node_id}`} · {t(`qualityGuard.eventTypes.${event.event}`)}

{t(`qualityGuard.reasons.${event.reason || "unknown"}`)}{event.output_tps ? ` · ${formatTPS(event.output_tps)}` : ""}

+

{event.node_name || `ID ${event.node_id}`} · {t(eventLabelKey(event.event))}

{t(reasonLabelKey(event.reason))}{event.account_id ? ` · ${t("qualityGuard.accountLease", { id: event.account_id })}` : ""}{event.request_id ? ` · ${event.request_id}` : ""}{event.cooldown_until ? ` · ${t("qualityGuard.leaseUntil", { time: formatTime(event.cooldown_until, locale) })}` : ""}{event.output_tps ? ` · ${formatTPS(event.output_tps)}` : ""}

)}
}
; } +function eventLabelKey(event: string): string { + if (event === "lease_scoped_quarantine_suppressed") return "qualityGuard.leaseScopedQuarantineSuppressedEvent"; + if (event === "lease_scoped_guard_released") return "qualityGuard.leaseScopedGuardReleasedEvent"; + if (event === "lease_quarantined") return "qualityGuard.leaseQuarantinedEvent"; + if (event === "lease_restored") return "qualityGuard.leaseRestoredEvent"; + if (event === "lease_quarantine_extended") return "qualityGuard.leaseQuarantineExtendedEvent"; + if (event === "lease_quarantine_failed" || event === "lease_quarantine_suppressed") return "qualityGuard.leaseQuarantineFailedEvent"; + return `qualityGuard.eventTypes.${event}`; +} + +function reasonLabelKey(reason: string): string { + if (reason === "lease_scoped_node") return "qualityGuard.leaseScopedNodeReason"; + if (reason === "fixed_fallback_node") return "qualityGuard.fixedFallback"; + return `qualityGuard.reasons.${reason || "unknown"}`; +} + function Policy({ status, onEdit }: { status: QualityGuardStatus; onEdit: () => void }) { const { t } = useTranslation(); const config = status.config; @@ -535,6 +555,7 @@ function qualityTestState(result: QualityTestResult, status: QualityGuardStatus) else if (result.outputTokensPerSecond >= softTPS) { classification = "soft"; reason = "soft_tps"; } const now = Date.now() / 1000; return { + observe_only: false, observe_only_reason: "", quarantined_lease_count: 0, active_soft_strikes: classification === "soft" ? 1 : classification === "hard" ? (status.config?.consecutive_soft ?? 2) : 0, passive_soft_strikes: 0, error_strikes: 0, quarantined_until: 0, disabled_by_guard: false, last_reason: reason, last_probe_at: now, last_observed_at: now, last_source: "active", diff --git a/frontend/src/shared/i18n/index.ts b/frontend/src/shared/i18n/index.ts index 8dea5ba8f..f1e07eee2 100644 --- a/frontend/src/shared/i18n/index.ts +++ b/frontend/src/shared/i18n/index.ts @@ -471,17 +471,32 @@ const resources = { }, qualityGuard: { title: "质量守护", description: "监测 Grok 出口质量并在异常时自动隔离节点。", overview: "质量守护概览", - serviceStatus: "守护服务", running: "运行正常", stale: "状态滞后", mode: "检测模式", availableNodes: "已启用节点", quarantinedNodes: "已隔离节点", + serviceStatus: "守护服务", running: "运行正常", stale: "状态滞后", mode: "检测模式", availableNodes: "已启用节点", quarantinedNodes: "已隔离节点", quarantinedTargets: "隔离对象", modes: { active: "主动检测", passive: "被动审计", hybrid: "混合模式" }, nodes: "节点质量", nodesHelp: "速度与 grok2api 面板同口径：输出 Token 包含推理 Token。首字后窗口短于首字等待且不足 1 秒时改用全程，避免加密思考被挤进最后几十毫秒。完整用户请求出现速度异常后会立即隔离，并在隔离期结束后使用受控探针复测。", updatedAt: "状态更新于 {{time}}", node: "节点", state: "状态", outputTPS: "面板输出速度", firstToken: "首字延迟", source: "数据来源", strikes: "打击计数", lastObserved: "最近观测", test: "检测", sources: { active: "主动探针", passive: "请求审计" }, quarantined: "已隔离", fixedFallback: "固定回退（受保护）", suspect: "可疑", healthy: "正常", pending: "待观测", probeFailed: "检测失败", - events: "最近事件", noEvents: "暂无异常或恢复事件", eventTypes: { node_quarantined: "节点已隔离", node_restored: "节点已恢复", node_rotated: "节点已更换 IP", passive_audit_anomaly: "检测到异常请求" }, + events: "最近事件", noEvents: "暂无异常或恢复事件", eventTypes: { node_quarantined: "节点已隔离", node_restored: "节点已恢复", node_rotated: "节点已更换 IP", passive_audit_anomaly: "检测到异常请求", lease_quarantined: "账号租约已隔离", lease_restored: "账号租约已恢复", lease_quarantine_extended: "账号租约隔离已延长", lease_quarantine_failed: "账号租约隔离未执行" }, statistics: "自动检测统计", statisticsSince: "自 {{time}} 开始累计，不含手动检测。", statisticsChecks: "有效检测总数", statisticsChecksHelp: "主动探测与有效被动审计", statisticsActive: "主动探测", statisticsActiveDetail: "正常 {{healthy}}，错误 {{errors}}", statisticsPassive: "被动审计", statisticsPassiveDetail: "正常 {{healthy}}，来自真实请求", statisticsTokens: "主动探测输出 Token", statisticsTokensHelp: "包含推理 Token，不代表代理流量", statisticsAnomalies: "异常命中", statisticsAnomalyDetail: "软异常 {{soft}}，硬异常 {{hard}}", statisticsQuarantines: "执行隔离", statisticsActionDetail: "已恢复 {{restored}}，受保护未隔离 {{suppressed}}", reasons: { unknown: "未记录原因", hard_tps: "超过硬阈值", soft_tps: "超过软阈值", buffered_burst: "短窗口输出突增，等待原 IP 复测", missing_thinking: "输出缺少 thinking", passive_hard_tps: "请求速度超过硬阈值", passive_soft_tps: "请求速度超过软阈值", quality_probe_healthy: "模型质量检测恢复正常", expected_marker_missing: "响应标记缺失", insufficient_output_tokens: "输出 Token 不足", insufficient_visible_tokens: "可见 Token 不足", insufficient_generation_window: "有效生成窗口不足", probe_errors: "主动检测连续失败", probe_no_account: "暂无可调度账号，已延后复测", recovery_probe_error: "恢复检测失败", rotation_error: "更换 IP 失败" }, policy: "当前策略", editPolicy: "编辑策略", editPolicyTitle: "编辑质量守护策略", editPolicyDescription: "保存后由守护进程热加载，无需重启服务。", restoreDefaults: "恢复默认值", policySaved: "策略已保存，正在热加载", invalidPolicyValue: "数值超出允许范围", softThresholdMustBeLower: "软阈值必须低于硬阈值", activeIntervalSeconds: "主动检测间隔（秒）", passiveIntervalSeconds: "被动审计间隔（秒）", consecutiveSoft: "主动软异常连续次数", consecutiveErrors: "检测错误连续次数", quarantineSeconds: "隔离时长（秒）", softThreshold: "软阈值", hardThreshold: "硬阈值", activeInterval: "主动间隔", passiveInterval: "审计间隔", quarantineDuration: "隔离时长", minimumNodes: "最少保留节点", unavailable: "质量守护尚未连接", unavailableHelp: "在 config.yaml 中启用 qualityGuard，并启动 quality-guard Compose profile 后，这里会显示实时状态。", testing: "正在检测节点质量", testComplete: "检测完成：{{speed}}", testFailed: "质量检测暂不可用，请稍后重试", nodesTab: "节点质量", + leaseScopedObserveOnly: "租约级（仅观测）", + leaseScoped: "租约级", + leaseQuarantined: "租约隔离 {{count}}", + leaseScopedHelp: "该节点按账号生成不同粘性租约；异常请求只隔离对应账号租约，不会停用整个共享节点。", + leaseScopedObserveOnlyHelp: "当前异常缺少账号身份或租约接口不可用，因此仅记录观测，不会停用整个共享节点。", + leaseQuarantinedEvent: "账号租约已隔离", + leaseRestoredEvent: "账号租约已恢复", + leaseQuarantineExtendedEvent: "账号租约隔离已延长", + leaseQuarantineFailedEvent: "账号租约隔离未执行", + leaseScopedQuarantineSuppressedEvent: "已阻止整节点隔离", + leaseScopedGuardReleasedEvent: "已解除旧版整节点隔离", + leaseScopedNodeReason: "节点包含多个账号粘性租约", + accountLease: "账号 {{id}}", + leaseUntil: "隔离至 {{time}}", + statisticsSuppressedActionDetail: "已恢复 {{restored}}，策略抑制 {{suppressed}}", profilesTab: "探针方案", profilesHelp: "主动质量探测用的 Prompt 与预期标记。标记缺失记为硬异常。", profileActive: "当前使用", @@ -1557,6 +1572,22 @@ const resources = { shell: { appearance: "Appearance", dark: "Dark", light: "Light", system: "System", language: "Language", navigation: "Navigation", openNavigation: "Open navigation" }, qualityGuard: { title: "Quality guard", description: "Monitor Grok egress quality and quarantine anomalous nodes automatically.", overview: "Quality guard overview", serviceStatus: "Guard service", running: "Running", stale: "Status stale", mode: "Detection mode", availableNodes: "Enabled nodes", quarantinedNodes: "Quarantined", modes: { active: "Active probes", passive: "Passive audits", hybrid: "Hybrid" }, nodes: "Node quality", nodesHelp: "Speed matches the grok2api panel: output tokens include reasoning tokens. A tail shorter than the first-token wait and under 1s uses the full request duration so encrypted thinking is not crushed into the flush. A completed user request with anomalous throughput is isolated immediately, then verified with a controlled probe after the hold.", updatedAt: "Updated {{time}}", node: "Node", state: "State", outputTPS: "Panel output speed", firstToken: "First token", source: "Source", strikes: "Strikes", lastObserved: "Last observed", test: "Test", sources: { active: "Active probe", passive: "Request audit" }, quarantined: "Quarantined", fixedFallback: "Fixed fallback (protected)", suspect: "Suspect", healthy: "Healthy", pending: "Pending", probeFailed: "Probe failed", events: "Recent events", noEvents: "No anomaly or recovery events", eventTypes: { node_quarantined: "Node quarantined", node_restored: "Node restored", node_rotated: "Node IP rotated", passive_audit_anomaly: "Anomalous request detected" }, statistics: "Automatic detection statistics", statisticsSince: "Accumulated since {{time}}. Manual tests are excluded.", statisticsChecks: "Valid checks", statisticsChecksHelp: "Active probes and valid passive audits", statisticsActive: "Active probes", statisticsActiveDetail: "Healthy {{healthy}}, errors {{errors}}", statisticsPassive: "Passive audits", statisticsPassiveDetail: "Healthy {{healthy}}, from real requests", statisticsTokens: "Active output tokens", statisticsTokensHelp: "Includes reasoning tokens; not proxy traffic", statisticsAnomalies: "Anomaly hits", statisticsAnomalyDetail: "Soft {{soft}}, hard {{hard}}", statisticsQuarantines: "Quarantines applied", statisticsActionDetail: "Restored {{restored}}, protected {{suppressed}}", reasons: { unknown: "No reason recorded", hard_tps: "Hard threshold exceeded", soft_tps: "Soft threshold exceeded", buffered_burst: "Short-window output burst; retesting the same IP", missing_thinking: "Missing thinking tokens", passive_hard_tps: "Request exceeded hard threshold", passive_soft_tps: "Request exceeded soft threshold", quality_probe_healthy: "Model quality probe recovered", expected_marker_missing: "Expected marker missing", insufficient_output_tokens: "Too few output tokens", insufficient_visible_tokens: "Too few visible tokens", insufficient_generation_window: "Generation window too short", probe_errors: "Repeated probe errors", probe_no_account: "No schedulable probe account; retry deferred", recovery_probe_error: "Recovery probe failed", rotation_error: "IP rotation failed" }, policy: "Current policy", editPolicy: "Edit policy", editPolicyTitle: "Edit quality guard policy", editPolicyDescription: "The guard hot-reloads saved changes without a service restart.", restoreDefaults: "Restore defaults", policySaved: "Policy saved and queued for hot reload", invalidPolicyValue: "Value is outside the allowed range", softThresholdMustBeLower: "The soft threshold must be lower than the hard threshold", activeIntervalSeconds: "Active interval (seconds)", passiveIntervalSeconds: "Passive interval (seconds)", consecutiveSoft: "Consecutive active soft strikes", consecutiveErrors: "Consecutive probe errors", quarantineSeconds: "Quarantine (seconds)", softThreshold: "Soft threshold", hardThreshold: "Hard threshold", activeInterval: "Active interval", passiveInterval: "Audit interval", quarantineDuration: "Quarantine", minimumNodes: "Minimum nodes", unavailable: "Quality guard is not connected", unavailableHelp: "Enable qualityGuard in config.yaml and start the quality-guard Compose profile to display live status here.", testing: "Testing node quality", testComplete: "Test complete: {{speed}}", testFailed: "Quality test is temporarily unavailable. Try again shortly.", refreshNodes: "Refresh nodes", nodeEnabled: "Node enabled", nodeDisabled: "Node disabled", nodesEnabled: "Selected nodes enabled", nodesDisabled: "Selected nodes disabled", enableNode: "Enable node {{name}}", disableNode: "Disable node {{name}}", nodeEditorDescription: "Manage Grok Build egress used by the quality guard. Proxy URLs are write-only; leave the field blank while editing to keep the current value.", nodeCapacityHelp: "Maximum number of bound accounts; 0 means unlimited.", deleteNodeTitle: "Delete proxy node?", deleteNodeDescription: "Node “{{name}}” will be permanently deleted. This action cannot be undone.", deleteNodesTitle: "Delete {{count}} selected nodes?", deleteNodesDescription: "The selected proxy nodes will be permanently deleted. This action cannot be undone.", nodesTab: "Node quality", + quarantinedTargets: "Quarantined targets", + leaseScopedObserveOnly: "Lease-scoped (observe only)", + leaseScoped: "Lease-scoped", + leaseQuarantined: "{{count}} lease quarantined", + leaseScopedHelp: "This node renders a different sticky lease per account. An anomalous request quarantines only that account lease, never the shared node.", + leaseScopedObserveOnlyHelp: "The anomaly lacks an account identity or the lease API is unavailable, so it is observed without disabling the shared node.", + leaseQuarantinedEvent: "Account lease quarantined", + leaseRestoredEvent: "Account lease restored", + leaseQuarantineExtendedEvent: "Account lease quarantine extended", + leaseQuarantineFailedEvent: "Account lease quarantine not applied", + leaseScopedQuarantineSuppressedEvent: "Whole-node quarantine prevented", + leaseScopedGuardReleasedEvent: "Legacy whole-node quarantine released", + leaseScopedNodeReason: "Node contains multiple account-specific sticky leases", + accountLease: "Account {{id}}", + leaseUntil: "Quarantined until {{time}}", + statisticsSuppressedActionDetail: "Restored {{restored}}, suppressed {{suppressed}}", profilesTab: "Probe profiles", profilesHelp: "Prompt and expected marker for active quality probes. A missing marker is a hard failure.", profileActive: "In use", diff --git a/tools/egress-quality-guard/README.md b/tools/egress-quality-guard/README.md index 2b37b06c5..7c5d254b5 100644 --- a/tools/egress-quality-guard/README.md +++ b/tools/egress-quality-guard/README.md @@ -43,6 +43,20 @@ your own traffic before allowing automatic quarantine. records a generic connectivity probe for diagnosis, then uses the real model-quality probe as the authority before re-enabling the node. +Account-bound proxy templates such as Resin usernames containing `{account}` +render a distinct sticky lease for each account. Scheduled node probes remain +suppressed because one lease cannot represent its siblings. A passive anomaly +removes only the audited account lease; after the hold, recovery pins a probe to +that same account and node, renews an unhealthy hold, and clears the durable +marker only with a matching CAS version. Routing stops enforcing a hold after +its deadline, so a stopped sidecar cannot strand an account indefinitely. +Rebinding an account atomically removes its old marker. If identity or the lease +API is unavailable, the guard falls back to observation and never disables the +shared node. Rendered proxy usernames and credentials never cross the API. +Lease reconciliation uses opaque keyset pagination and scans the complete +durable set. Recovery probes are capped per cycle and retry with exponential +backoff, so a large expired queue cannot monopolize one guard cycle. + The public inference API cannot request a specific egress node or bypass a disabled node. This capability is confined to the authenticated internal route. Ambiguous probe-only 403 responses do not cool borrowed accounts; definitive @@ -112,6 +126,10 @@ probe prompt, or model response body. - Never deletes a node or changes account bindings. - Never restores a node disabled by an operator. +- Never applies whole-node quarantine to an account-bound `{account}` proxy. A + legacy quarantine still owned by the guard is released during reconciliation. +- Lease recovery is pinned to the same account and node and uses an opaque CAS + version so stale probes cannot clear a newer quarantine. - Refuses to quarantine below `qualityGuard.minimumHealthyNodes`. - Strict mode overrides that floor rather than scheduling an unverified exit. - Uses an exclusive process lock to prevent duplicate guards. diff --git a/tools/egress-quality-guard/README.zh-CN.md b/tools/egress-quality-guard/README.zh-CN.md index 6f4b07b14..2e8d2e239 100644 --- a/tools/egress-quality-guard/README.zh-CN.md +++ b/tools/egress-quality-guard/README.zh-CN.md @@ -23,6 +23,15 @@ Token/s，因此建议先观察 JSON 日志，再根据实际流量调整阈值 6. 隔离节点仍可接受管理员探测，但不会承载普通用户请求。 7. 冷却结束后记录一次通用连接探测用于诊断，再以真实模型质量探测作为恢复判据，账号绑定保持不变。 +Resin 用户名等代理模板包含 `{account}` 时，同一逻辑节点会按账号生成不同的粘性租约。 +这类节点不会执行无法代表全部租约的定时节点探测；被动审计出现异常时，后端只临时移出该 +审计关联的账号租约。冷却到期后，恢复探针固定使用同一账号与同一节点；异常会续期，健康时 +通过版本校验清理持久化标记。路由在隔离期限到期后不再强制摘流，避免 sidecar 停止时让孤儿 +状态永久卡住账号。账号换绑会原子清理旧标记；sidecar 或接口异常时只记录观测，绝不回退为 +整节点禁用。内部接口只传账号 ID、节点 ID 和随机版本号，不返回渲染后的 Resin 用户名或代理凭据。 +租约对账采用不透明游标完整分页；每轮恢复探针有固定上限，失败后按指数退避，避免大量到期租约 +长期占满单次守护循环。 + 普通 `/v1/*` 请求不能指定出口节点，也不能绕过节点禁用状态。 仅发生在质量探测中的模糊 403 不会冷却借用账号；明确的凭据失效、账号封禁和额度信号仍按原有规则处理。 @@ -70,6 +79,7 @@ Webhook，确认出口发生变化，再执行一次真实模型质量检测； - 不删除节点，不修改账号绑定。 - 不会恢复管理员手动禁用的节点。 +- 不会对账号绑定的 `{account}` 代理执行整节点隔离；升级前仍由守护程序持有的旧隔离会在状态对账时解除。 - 启用节点数低于 `qualityGuard.minimumHealthyNodes` 时拒绝继续隔离。 - 严格模式会覆盖最低健康节点保护：无法确认质量时宁可无可用节点，也不调度可疑出口。 - 使用进程锁防止重复运行。 diff --git a/tools/egress-quality-guard/quality_guard.py b/tools/egress-quality-guard/quality_guard.py index 584d10b6a..0eb25be73 100755 --- a/tools/egress-quality-guard/quality_guard.py +++ b/tools/egress-quality-guard/quality_guard.py @@ -47,6 +47,11 @@ QUALITY_MARKER_PROFILE_ID = "quality-marker" THROUGHPUT_PROFILE_ID = "throughput" THINKING_GUARD_MIN_OUTPUT_TOKENS = 64 +LEASE_PAGE_SIZE = 1000 +LEASE_SCAN_MAX_PAGES = 1000 +LEASE_RECOVERY_MAX_PER_CYCLE = 8 +LEASE_RECOVERY_BACKOFF_BASE_SECONDS = 30 +LEASE_RECOVERY_BACKOFF_MAX_SECONDS = 1800 class GuardDisabled(RuntimeError): @@ -321,8 +326,10 @@ def fixed_fallback_node_ids(self) -> set[str]: result.add(node_id) return result - def quality_test(self, node_id: str, profile_id: str = "") -> dict[str, Any]: + def quality_test(self, node_id: str, profile_id: str = "", account_id: str = "") -> dict[str, Any]: body = {"profileId": profile_id} if profile_id else {} + if account_id: + body["accountId"] = account_id return self._request("POST", f"{INTERNAL_API_PREFIX}/egress-nodes/{node_id}/quality-test", body or None) def connectivity_test(self, node_id: str) -> dict[str, Any]: @@ -342,6 +349,42 @@ def set_enabled(self, node_id: str, enabled: bool) -> int: result = self._request("PATCH", f"{INTERNAL_API_PREFIX}/egress-nodes/batch", {"ids": [node_id], "enabled": enabled}) return int(result.get("updated") or 0) + def list_leases(self) -> list[dict[str, Any]]: + values: list[dict[str, Any]] = [] + cursor = "" + seen_cursors: set[str] = set() + for _page in range(LEASE_SCAN_MAX_PAGES): + query = {"limit": LEASE_PAGE_SIZE} + if cursor: + query["cursor"] = cursor + payload = self._request("GET", f"{INTERNAL_API_PREFIX}/egress-leases?{urllib.parse.urlencode(query)}") + items = list(payload.get("items") or []) + values.extend(items) + if not payload.get("hasMore"): + return values + next_cursor = str(payload.get("nextCursor") or "") + if not next_cursor or next_cursor == cursor or next_cursor in seen_cursors: + raise RuntimeError("lease pagination did not advance") + seen_cursors.add(next_cursor) + cursor = next_cursor + raise RuntimeError("lease pagination exceeded the safety limit") + + def quarantine_lease(self, node_id: str, account_id: str, reason: str) -> dict[str, Any]: + return self._request("POST", f"{INTERNAL_API_PREFIX}/egress-leases/quarantine", { + "nodeId": node_id, + "accountId": account_id, + "reason": reason, + "quarantineSeconds": self.config.quarantine_seconds, + }) + + def restore_lease(self, node_id: str, account_id: str, version: str) -> bool: + result = self._request("POST", f"{INTERNAL_API_PREFIX}/egress-leases/restore", { + "nodeId": node_id, + "accountId": account_id, + "version": version, + }) + return bool(result.get("restored")) + def rotate_node(self, node_id: str, old_exit_ip: str = "") -> dict[str, Any]: if not self.config.rotation_url: raise RuntimeError("rotation endpoint is not configured") @@ -517,6 +560,9 @@ def generation_window_ms(first_token_ms: int, duration_ms: int, reasoning_tokens def default_node_state() -> dict[str, Any]: return { + "observe_only": False, + "observe_only_reason": "", + "quarantined_lease_count": 0, "active_soft_strikes": 0, "passive_soft_strikes": 0, "error_strikes": 0, @@ -630,6 +676,8 @@ def __init__(self, config: Config, api: ApiClient): self._resolved_node_ids = list(config.node_ids) self.state.setdefault("started_at", time.time()) self.state.setdefault("recent_events", []) + self.state.setdefault("leases", {}) + self.state.setdefault("lease_recovery", {}) ensure_statistics(self.state) self._update_guard_metadata() self._save() @@ -678,6 +726,11 @@ def _state_for(self, node_id: str) -> dict[str, Any]: current.setdefault(key, value) return current + @staticmethod + def _is_lease_scoped(node: dict[str, Any]) -> bool: + """Return whether one logical node expands to account-specific sticky leases.""" + return bool(node.get("accountBoundProxy")) + def _defer_no_account(self, state: dict[str, Any], node: dict[str, Any], now: float, event: str, **fields: Any) -> None: state["last_probe_at"] = now state["last_reason"] = "probe_no_account" @@ -732,9 +785,213 @@ def _should_rotate(self, node_id: str, reason: str) -> bool: def _probe_account_unavailable(exc: Exception) -> bool: return isinstance(exc, ApiError) and exc.code == "egressQualityProbeNoAccount" + @staticmethod + def _lease_key(node_id: str, account_id: str) -> str: + return f"{node_id}:{account_id}" + + def _clear_lease_recovery(self, key: str) -> None: + self.state.setdefault("lease_recovery", {}).pop(key, None) + + def _defer_lease_recovery(self, key: str, now: float) -> None: + recovery = self.state.setdefault("lease_recovery", {}) + current = recovery.get(key) or {} + failures = min(16, int(current.get("failures") or 0) + 1) + delay = min(LEASE_RECOVERY_BACKOFF_MAX_SECONDS, LEASE_RECOVERY_BACKOFF_BASE_SECONDS * (2 ** (failures - 1))) + recovery[key] = {"failures": failures, "next_attempt_at": now + delay} + + def _quarantine_lease(self, node: dict[str, Any], audit_value: dict[str, Any], reason: str, now: float) -> None: + node_id = str(node.get("id") or "") + account_id = str(audit_value.get("accountId") or "") + state = self._state_for(node_id) + request_id = str(audit_value.get("requestId") or "") + if not account_id: + state.update({"observe_only": True, "observe_only_reason": "missing_account_identity", "last_reason": reason}) + self._bump_statistic("actions", "suppressed") + append_state_event(self.state, "lease_quarantine_suppressed", node_id=node_id, node_name=node.get("name"), reason=reason, request_id=request_id) + log_event("lease_quarantine_suppressed", node_id=node_id, node_name=node.get("name"), reason=reason, cause="missing_account_identity") + return + try: + lease = self.api.quarantine_lease(node_id, account_id, reason) + except Exception as exc: + state.update({"observe_only": True, "observe_only_reason": "lease_api_unavailable", "last_reason": reason}) + self._bump_statistic("actions", "suppressed") + append_state_event(self.state, "lease_quarantine_failed", node_id=node_id, node_name=node.get("name"), reason=reason, account_id=account_id, request_id=request_id) + log_event("lease_quarantine_failed", node_id=node_id, node_name=node.get("name"), reason=reason, error_type=type(exc).__name__) + return + key = self._lease_key(node_id, account_id) + leases = self.state.setdefault("leases", {}) + already_quarantined = key in leases + leases[key] = lease + self._clear_lease_recovery(key) + state.update({"observe_only": False, "observe_only_reason": "", "last_reason": reason}) + event = "lease_quarantine_extended" if already_quarantined else "lease_quarantined" + if not already_quarantined: + state["quarantined_lease_count"] = int(state.get("quarantined_lease_count", 0)) + 1 + self._bump_statistic("actions", "quarantined") + append_state_event(self.state, event, node_id=node_id, node_name=node.get("name"), reason=reason, account_id=account_id, request_id=request_id, cooldown_until=lease.get("cooldownUntil")) + self._save() + log_event(event, node_id=node_id, node_name=node.get("name"), reason=reason, account_id=account_id, cooldown_until=lease.get("cooldownUntil")) + + def _extend_lease(self, node: dict[str, Any], lease: dict[str, Any], reason: str) -> bool: + node_id = str(lease.get("nodeId") or node.get("id") or "") + account_id = str(lease.get("accountId") or "") + try: + replacement = self.api.quarantine_lease(node_id, account_id, reason) + except Exception as exc: + log_event("lease_quarantine_extension_failed", node_id=node_id, node_name=node.get("name"), reason=reason, error_type=type(exc).__name__) + return False + key = self._lease_key(node_id, account_id) + self.state.setdefault("leases", {})[key] = replacement + self._clear_lease_recovery(key) + append_state_event(self.state, "lease_quarantine_extended", node_id=node_id, node_name=node.get("name"), reason=reason, account_id=account_id, cooldown_until=replacement.get("cooldownUntil")) + log_event("lease_quarantine_extended", node_id=node_id, node_name=node.get("name"), reason=reason, account_id=account_id) + return True + + def _recover_lease(self, node: dict[str, Any], lease: dict[str, Any], now: float) -> None: + node_id = str(lease.get("nodeId") or "") + account_id = str(lease.get("accountId") or "") + version = str(lease.get("version") or "") + if not node_id or not account_id or not version: + return + key = self._lease_key(node_id, account_id) + profile_id, profile = resolve_probe_profile(self.config.profiles_file, QUALITY_MARKER_PROFILE_ID) + self._bump_statistic("active", "total") + try: + result = self.api.quality_test(node_id, profile_id, account_id) + classification, reason = classify_result(result, self.config, profile) + except Exception as exc: + self._bump_statistic("active", "errors") + if not self._extend_lease(node, lease, "recovery_probe_error"): + self._defer_lease_recovery(key, now) + log_event("lease_recovery_probe_failed", node_id=node_id, node_name=node.get("name"), account_id=account_id, error_type=type(exc).__name__) + return + self._record_probe(node, result, classification, reason, now) + if classification != "healthy": + if not self._extend_lease(node, lease, reason): + self._defer_lease_recovery(key, now) + return + try: + restored = self.api.restore_lease(node_id, account_id, version) + except ApiError as exc: + if exc.code == "qualityLeaseConflict": + self._clear_lease_recovery(key) + log_event("lease_restore_stale", node_id=node_id, node_name=node.get("name"), account_id=account_id) + return + self._defer_lease_recovery(key, now) + log_event("lease_restore_failed", node_id=node_id, node_name=node.get("name"), account_id=account_id, error_type=type(exc).__name__) + return + except Exception as exc: + self._defer_lease_recovery(key, now) + log_event("lease_restore_failed", node_id=node_id, node_name=node.get("name"), account_id=account_id, error_type=type(exc).__name__) + return + if not restored: + self._defer_lease_recovery(key, now) + return + self.state.setdefault("leases", {}).pop(key, None) + self._clear_lease_recovery(key) + node_state = self._state_for(node_id) + node_state["quarantined_lease_count"] = max(0, int(node_state.get("quarantined_lease_count", 0)) - 1) + self._bump_statistic("actions", "restored") + append_state_event(self.state, "lease_restored", node_id=node_id, node_name=node.get("name"), reason="quality_probe_healthy", account_id=account_id) + log_event("lease_restored", node_id=node_id, node_name=node.get("name"), account_id=account_id, reason="quality_probe_healthy") + + def _reconcile_leases(self, nodes: list[dict[str, Any]], now: float) -> bool: + node_by_id = {str(node.get("id") or ""): node for node in nodes} + try: + values = self.api.list_leases() + except Exception as exc: + log_event("lease_reconciliation_failed", error_type=type(exc).__name__) + return False + state_leases = self.state.setdefault("leases", {}) + recovery_state = self.state.setdefault("lease_recovery", {}) + backend_keys: set[str] = set() + due: list[tuple[dict[str, Any], dict[str, Any]]] = [] + for node in nodes: + if self._is_lease_scoped(node): + self._state_for(str(node["id"]))["quarantined_lease_count"] = 0 + for lease in values: + node_id = str(lease.get("nodeId") or "") + account_id = str(lease.get("accountId") or "") + if not node_id or not account_id: + continue + key = self._lease_key(node_id, account_id) + backend_keys.add(key) + state_leases[key] = lease + node = node_by_id.get(node_id) + if node is None or not self._is_lease_scoped(node): + continue + state = self._state_for(node_id) + state["observe_only"] = False + state["observe_only_reason"] = "" + state["quarantined_lease_count"] = int(state.get("quarantined_lease_count", 0)) + 1 + if now >= float(lease.get("cooldownUntil") or 0): + retry = recovery_state.get(key) or {} + if now >= float(retry.get("next_attempt_at") or 0): + due.append((node, lease)) + for key in list(state_leases): + if key not in backend_keys: + state_leases.pop(key, None) + recovery_state.pop(key, None) + for key in list(recovery_state): + if key not in backend_keys: + recovery_state.pop(key, None) + for node, lease in due[:LEASE_RECOVERY_MAX_PER_CYCLE]: + self._recover_lease(node, lease, now) + deferred = max(0, len(due) - LEASE_RECOVERY_MAX_PER_CYCLE) + if deferred: + log_event("lease_recovery_budget_exhausted", due=len(due), deferred=deferred, limit=LEASE_RECOVERY_MAX_PER_CYCLE) + return True + + def _release_protected_leases(self, protected_node_ids: set[str], nodes: list[dict[str, Any]]) -> None: + if not protected_node_ids: + return + node_by_id = {str(node.get("id") or ""): node for node in nodes} + state_leases = self.state.setdefault("leases", {}) + for key, lease in list(state_leases.items()): + node_id = str(lease.get("nodeId") or "") + if node_id not in protected_node_ids: + continue + account_id = str(lease.get("accountId") or "") + version = str(lease.get("version") or "") + try: + restored = self.api.restore_lease(node_id, account_id, version) + except Exception as exc: + log_event("protected_lease_release_failed", node_id=node_id, error_type=type(exc).__name__) + continue + if not restored: + continue + state_leases.pop(key, None) + node = node_by_id.get(node_id) or {} + state = self._state_for(node_id) + state["quarantined_lease_count"] = max(0, int(state.get("quarantined_lease_count", 0)) - 1) + self._bump_statistic("actions", "restored") + append_state_event(self.state, "lease_restored", node_id=node_id, node_name=node.get("name"), reason="fixed_fallback_node", account_id=account_id) + log_event("protected_lease_released", node_id=node_id, node_name=node.get("name"), account_id=account_id) + def _quarantine(self, nodes: list[dict[str, Any]], node: dict[str, Any], reason: str, now: float, recover_now: bool = True) -> None: node_id = str(node["id"]) state = self._state_for(node_id) + if self._is_lease_scoped(node): + state.update({ + "observe_only": True, + "observe_only_reason": "account_bound_proxy", + "last_reason": reason, + }) + self._bump_statistic("actions", "suppressed") + append_state_event( + self.state, + "lease_scoped_quarantine_suppressed", + node_id=node_id, + node_name=node.get("name"), + reason=reason, + ) + log_event( + "lease_scoped_quarantine_suppressed", + node_id=node_id, + node_name=node.get("name"), + reason=reason, + ) + return if not self._can_quarantine(nodes, node_id): self._bump_statistic("actions", "suppressed") log_event("quarantine_suppressed", node_id=node_id, node_name=node.get("name"), reason=reason, minimum_healthy=self.config.min_healthy_nodes) @@ -993,13 +1250,80 @@ def _probe_quarantined(self, node: dict[str, Any], now: float) -> None: def _prepare_nodes(self, now: float) -> tuple[list[dict[str, Any]], list[dict[str, Any]], set[str]]: all_nodes = self.api.list_nodes() + lease_api_ready = self._reconcile_leases(all_nodes, now) protected_node_ids = self.api.fixed_fallback_node_ids() + if lease_api_ready: + self._release_protected_leases(protected_node_ids, all_nodes) previous_protected = set(str(value) for value in self.state.get("protected_node_ids", [])) if protected_node_ids != previous_protected: self.state["protected_node_ids"] = sorted(protected_node_ids) for node_id in sorted(protected_node_ids - previous_protected): log_event("fixed_fallback_node_skipped", node_id=node_id) state_nodes = self.state.setdefault("nodes", {}) + release_failed_ids: set[str] = set() + # An account-bound proxy renders a different sticky lease for each + # account. Release any whole-node quarantine left by an older guard; + # current versions isolate the audited account lease instead. + for node in all_nodes: + node_id = str(node.get("id") or "") + if not node_id or not node.get("proxyConfigured"): + continue + existing = state_nodes.get(node_id) + if not self._is_lease_scoped(node): + if existing: + existing["observe_only"] = False + existing["observe_only_reason"] = "" + continue + state = self._state_for(node_id) + if not lease_api_ready: + state["observe_only"] = True + state["observe_only_reason"] = "lease_api_unavailable" + elif state.get("observe_only_reason") in {"account_bound_proxy", "lease_api_unavailable"}: + state["observe_only"] = False + state["observe_only_reason"] = "" + if not state.get("disabled_by_guard"): + continue + if not node.get("enabled"): + try: + updated = self.api.set_enabled(node_id, True) + except Exception as exc: + release_failed_ids.add(node_id) + log_event( + "lease_scoped_guard_release_failed", + node_id=node_id, + node_name=node.get("name"), + error_type=type(exc).__name__, + ) + continue + if updated != 1: + release_failed_ids.add(node_id) + log_event( + "lease_scoped_guard_release_not_applied", + node_id=node_id, + node_name=node.get("name"), + updated=updated, + ) + continue + node["enabled"] = True + self._bump_statistic("actions", "restored") + state.update({ + "active_soft_strikes": 0, + "passive_soft_strikes": 0, + "error_strikes": 0, + "quarantined_until": 0.0, + "disabled_by_guard": False, + "last_reason": "", + "quarantine_source": "", + }) + append_state_event( + self.state, + "lease_scoped_guard_released", + node_id=node_id, + node_name=node.get("name"), + reason="lease_scoped_node", + ) + self._save() + log_event("lease_scoped_guard_released", node_id=node_id, node_name=node.get("name")) # Making an enabled node a fixed fallback is an explicit operator # override. Relinquish stale guard ownership before eligibility checks # so strict mode cannot repeatedly attempt an invalid disable. A @@ -1030,7 +1354,7 @@ def _prepare_nodes(self, now: float) -> tuple[list[dict[str, Any]], list[dict[st tracked = bool((state_nodes.get(stale_id) or {}).get("disabled_by_guard")) if stale_id not in present_ids or (stale_id not in managed_ids and not tracked): del state_nodes[stale_id] - skip_ids: set[str] = set() + skip_ids: set[str] = set(release_failed_ids) if not nodes: log_event("no_eligible_nodes") return all_nodes, [], skip_ids @@ -1067,6 +1391,8 @@ def _prepare_nodes(self, now: float) -> tuple[list[dict[str, Any]], list[dict[st continue if state.get("disabled_by_guard"): skip_ids.add(node_id) + if self._is_lease_scoped(node): + continue self._probe_quarantined(node, now) return all_nodes, nodes, skip_ids @@ -1076,6 +1402,9 @@ def run_active_cycle(self) -> None: for node in nodes: node_id = str(node["id"]) state = self._state_for(node_id) + if self._is_lease_scoped(node): + self._save() + continue if node_id not in skip_ids and node.get("enabled") and not state.get("disabled_by_guard"): self._probe_active(all_nodes, node, now) self._save() @@ -1175,15 +1504,18 @@ def _record_passive_audit(self, all_nodes: list[dict[str, Any]], node: dict[str, duration_ms=int(audit_value.get("durationMs") or 0), strikes=int(state.get("passive_soft_strikes", 0)), ) - log_event( - "passive_immediate_quarantine", - node_id=node_id, - node_name=node.get("name"), - classification=classification, - reason=reason, - output_tps=round(speed, 3), - ) - self._quarantine(all_nodes, node, reason, now, recover_now=False) + if self._is_lease_scoped(node): + self._quarantine_lease(node, audit_value, reason, now) + else: + log_event( + "passive_immediate_quarantine", + node_id=node_id, + node_name=node.get("name"), + classification=classification, + reason=reason, + output_tps=round(speed, 3), + ) + self._quarantine(all_nodes, node, reason, now, recover_now=False) def run_passive_cycle(self) -> None: now = time.time() diff --git a/tools/egress-quality-guard/quality_guard_test.py b/tools/egress-quality-guard/quality_guard_test.py index d6b574b76..e6ef7d9c0 100644 --- a/tools/egress-quality-guard/quality_guard_test.py +++ b/tools/egress-quality-guard/quality_guard_test.py @@ -342,6 +342,28 @@ def test_fixed_fallback_nodes_are_discovered_from_operations_policy(self): } self.assertEqual(client.fixed_fallback_node_ids(), {"9", "11"}) + def test_list_leases_uses_stable_cursor_until_complete(self): + client = quality_guard.ApiClient(config()) + requested_cursors = [] + + def request(_method, path, _body=None): + query = quality_guard.urllib.parse.parse_qs(quality_guard.urllib.parse.urlparse(path).query) + cursor = (query.get("cursor") or [""])[0] + requested_cursors.append(cursor) + if not cursor: + return {"items": [{"accountId": "1"}], "hasMore": True, "nextCursor": "next-page"} + return {"items": [{"accountId": "2"}], "hasMore": False, "nextCursor": ""} + + client._request = request + self.assertEqual([value["accountId"] for value in client.list_leases()], ["1", "2"]) + self.assertEqual(requested_cursors, ["", "next-page"]) + + def test_list_leases_rejects_non_advancing_cursor(self): + client = quality_guard.ApiClient(config()) + client._request = lambda *_args, **_kwargs: {"items": [], "hasMore": True, "nextCursor": "same"} + with self.assertRaises(RuntimeError): + client.list_leases() + class FakeApi: def __init__(self, nodes, results, audit_pages=None, fixed_fallback_ids=None): @@ -352,7 +374,9 @@ def __init__(self, nodes, results, audit_pages=None, fixed_fallback_ids=None): self.enabled_calls = [] self.quality_calls = [] self.quality_profile_calls = [] + self.quality_account_calls = [] self.rotation_calls = [] + self.leases = {} def list_nodes(self): return self.nodes @@ -360,9 +384,10 @@ def list_nodes(self): def fixed_fallback_node_ids(self): return set(self.fixed_fallback_ids) - def quality_test(self, node_id, profile_id=""): + def quality_test(self, node_id, profile_id="", account_id=""): self.quality_calls.append(node_id) self.quality_profile_calls.append(profile_id) + self.quality_account_calls.append(account_id) value = self.results.pop(0) if isinstance(value, Exception): raise value @@ -388,6 +413,26 @@ def list_audits(self, _cursor=""): return self.audit_pages.pop(0) return {"items": [], "hasMore": False, "nextCursor": ""} + def list_leases(self): + return list(self.leases.values()) + + def quarantine_lease(self, node_id, account_id, reason): + key = f"{node_id}:{account_id}" + value = { + "nodeId": node_id, "accountId": account_id, "reason": reason, + "version": f"version-{len(self.leases) + 1:09d}", "cooldownUntil": time.time() + 300, + } + self.leases[key] = value + return value + + def restore_lease(self, node_id, account_id, version): + key = f"{node_id}:{account_id}" + current = self.leases.get(key) + if current is None or current.get("version") != version: + raise quality_guard.ApiError(409, "qualityLeaseConflict", "stale") + del self.leases[key] + return True + class GuardTests(unittest.TestCase): @staticmethod @@ -444,6 +489,182 @@ def test_fixed_fallback_node_is_excluded_without_aborting_other_nodes(self): self.assertEqual(api.quality_calls, ["2"]) self.assertEqual(guard.state["protected_node_ids"], ["1"]) + def test_fixed_fallback_releases_existing_account_lease_without_probe(self): + with tempfile.TemporaryDirectory() as directory: + cfg = config(state_file=Path(directory) / "state.json", lock_file=Path(directory) / "lock", node_ids=("1",)) + nodes = self.nodes(2) + nodes[0]["accountBoundProxy"] = True + api = FakeApi(nodes, [], fixed_fallback_ids={"1"}) + api.leases["1:101"] = { + "nodeId": "1", "accountId": "101", "reason": "hard_tps", + "version": "lease-version-0001", "cooldownUntil": time.time() + 300, + } + guard = quality_guard.Guard(cfg, api) + guard.run_active_cycle() + self.assertEqual(api.leases, {}) + self.assertEqual(api.quality_calls, []) + self.assertEqual(guard.state["statistics"]["actions"]["restored"], 1) + self.assertEqual(guard.state["recent_events"][-1]["event"], "lease_restored") + + def test_account_bound_proxy_skips_scheduled_probe_and_stays_observable(self): + with tempfile.TemporaryDirectory() as directory: + cfg = config( + state_file=Path(directory) / "state.json", + lock_file=Path(directory) / "lock", + node_ids=("1",), + ) + nodes = self.nodes(3) + nodes[0]["accountBoundProxy"] = True + api = FakeApi(nodes, []) + guard = quality_guard.Guard(cfg, api) + guard.run_active_cycle() + + self.assertEqual(api.quality_calls, []) + self.assertEqual(api.enabled_calls, []) + self.assertTrue(nodes[0]["enabled"]) + self.assertFalse(guard.state["nodes"]["1"]["observe_only"]) + self.assertEqual(guard.state["nodes"]["1"]["observe_only_reason"], "") + + def test_account_bound_proxy_records_passive_anomaly_without_node_quarantine(self): + with tempfile.TemporaryDirectory() as directory: + cfg = config( + state_file=Path(directory) / "state.json", + lock_file=Path(directory) / "lock", + mode="passive", + node_ids=("1",), + ) + nodes = self.nodes(3) + nodes[0]["accountBoundProxy"] = True + api = FakeApi(nodes, [{"expectedMatched": True, "outputTokens": 100, "reasoningTokens": 40, "outputTokensPerSecond": 100}], [ + {"items": [], "hasMore": False, "nextCursor": ""}, + {"items": [self.audit("lease-hit", "1", 1200)], "hasMore": False, "nextCursor": ""}, + ]) + guard = quality_guard.Guard(cfg, api) + guard.run_passive_cycle() + guard.run_passive_cycle() + + state = guard.state["nodes"]["1"] + self.assertEqual(api.enabled_calls, []) + self.assertTrue(nodes[0]["enabled"]) + self.assertFalse(state["disabled_by_guard"]) + self.assertEqual(state["last_classification"], "hard") + self.assertEqual(state["last_reason"], "hard_tps") + self.assertEqual(guard.state["statistics"]["actions"]["quarantined"], 1) + self.assertEqual(len(api.leases), 1) + self.assertEqual( + [event["event"] for event in guard.state["recent_events"]], + ["passive_audit_anomaly", "lease_quarantined"], + ) + next(iter(api.leases.values()))["cooldownUntil"] = 0 + guard.run_active_cycle() + self.assertEqual(api.quality_account_calls, ["101"]) + self.assertEqual(api.leases, {}) + self.assertEqual(guard.state["recent_events"][-1]["event"], "lease_restored") + + def test_repeated_account_anomaly_extends_one_lease_without_double_counting(self): + with tempfile.TemporaryDirectory() as directory: + cfg = config(state_file=Path(directory) / "state.json", lock_file=Path(directory) / "lock", node_ids=("1",)) + nodes = self.nodes(2) + nodes[0]["accountBoundProxy"] = True + api = FakeApi(nodes, []) + guard = quality_guard.Guard(cfg, api) + audit = {"accountId": "101"} + + guard._quarantine_lease(nodes[0], audit, "hard_tps", time.time()) + guard._quarantine_lease(nodes[0], audit, "hard_tps", time.time()) + + self.assertEqual(len(api.leases), 1) + self.assertEqual(guard.state["nodes"]["1"]["quarantined_lease_count"], 1) + self.assertEqual(guard.state["statistics"]["actions"]["quarantined"], 1) + self.assertEqual(guard.state["recent_events"][-1]["event"], "lease_quarantine_extended") + + def test_due_lease_recovery_has_a_per_cycle_budget(self): + with tempfile.TemporaryDirectory() as directory: + cfg = config(state_file=Path(directory) / "state.json", lock_file=Path(directory) / "lock", node_ids=("1",)) + nodes = self.nodes(2) + nodes[0]["accountBoundProxy"] = True + healthy = {"expectedMatched": True, "outputTokens": 100, "reasoningTokens": 40, "outputTokensPerSecond": 100} + api = FakeApi(nodes, [healthy] * 20) + for index in range(20): + account_id = str(1000 + index) + api.leases[f"1:{account_id}"] = { + "nodeId": "1", "accountId": account_id, "reason": "hard_tps", + "version": f"lease-version-{index:04d}", "cooldownUntil": 0, + } + guard = quality_guard.Guard(cfg, api) + guard.run_active_cycle() + self.assertEqual(len(api.quality_account_calls), quality_guard.LEASE_RECOVERY_MAX_PER_CYCLE) + self.assertEqual(len(api.leases), 20 - quality_guard.LEASE_RECOVERY_MAX_PER_CYCLE) + guard.run_active_cycle() + self.assertEqual(len(api.quality_account_calls), quality_guard.LEASE_RECOVERY_MAX_PER_CYCLE * 2) + + def test_failed_lease_recovery_is_backed_off(self): + with tempfile.TemporaryDirectory() as directory: + cfg = config(state_file=Path(directory) / "state.json", lock_file=Path(directory) / "lock", node_ids=("1",)) + nodes = self.nodes(2) + nodes[0]["accountBoundProxy"] = True + api = FakeApi(nodes, [RuntimeError("probe failed"), RuntimeError("must not run immediately")]) + api.leases["1:101"] = { + "nodeId": "1", "accountId": "101", "reason": "hard_tps", + "version": "lease-version-0001", "cooldownUntil": 0, + } + api.quarantine_lease = mock.Mock(side_effect=RuntimeError("backend unavailable")) + guard = quality_guard.Guard(cfg, api) + guard.run_active_cycle() + guard.run_active_cycle() + self.assertEqual(api.quality_account_calls, ["101"]) + retry = guard.state["lease_recovery"]["1:101"] + self.assertGreater(retry["next_attempt_at"], time.time()) + + def test_account_bound_proxy_releases_only_guard_owned_legacy_quarantine(self): + with tempfile.TemporaryDirectory() as directory: + cfg = config( + state_file=Path(directory) / "state.json", + lock_file=Path(directory) / "lock", + node_ids=("1",), + ) + nodes = self.nodes(3) + nodes[0].update({"accountBoundProxy": True, "enabled": False}) + api = FakeApi(nodes, []) + guard = quality_guard.Guard(cfg, api) + state = guard._state_for("1") + state.update({"disabled_by_guard": True, "last_reason": "hard_tps", "quarantined_until": time.time() + 300}) + guard.run_active_cycle() + + self.assertEqual(api.enabled_calls, [("1", True)]) + self.assertTrue(nodes[0]["enabled"]) + self.assertFalse(state["disabled_by_guard"]) + self.assertFalse(state["observe_only"]) + self.assertEqual(guard.state["statistics"]["actions"]["restored"], 1) + self.assertEqual(guard.state["recent_events"][-1]["event"], "lease_scoped_guard_released") + + nodes[0]["enabled"] = False + guard.run_active_cycle() + self.assertEqual(api.enabled_calls, [("1", True)]) + self.assertFalse(nodes[0]["enabled"]) + + def test_account_bound_proxy_keeps_ownership_when_legacy_release_fails(self): + with tempfile.TemporaryDirectory() as directory: + cfg = config( + state_file=Path(directory) / "state.json", + lock_file=Path(directory) / "lock", + node_ids=("1",), + ) + nodes = self.nodes(3) + nodes[0].update({"accountBoundProxy": True, "enabled": False}) + api = FakeApi(nodes, []) + api.set_enabled = mock.Mock(side_effect=RuntimeError("temporary backend failure")) + guard = quality_guard.Guard(cfg, api) + state = guard._state_for("1") + state.update({"disabled_by_guard": True, "last_reason": "hard_tps", "quarantined_until": 0}) + guard.run_active_cycle() + + api.set_enabled.assert_called_once_with("1", True) + self.assertEqual(api.quality_calls, []) + self.assertFalse(nodes[0]["enabled"]) + self.assertTrue(state["disabled_by_guard"]) + self.assertFalse(state["observe_only"]) + def test_enabled_node_promoted_to_fixed_fallback_releases_guard_ownership(self): with tempfile.TemporaryDirectory() as directory: cfg = config( @@ -990,7 +1211,7 @@ def audit(audit_id, node_id, output_tps, quality_probe=False): "provider": "grok_build", "streaming": True, "statusCode": 200, "firstTokenMs": 200, "durationMs": 200 + generation_ms, "outputTokens": output_tokens, "reasoningTokens": min(100, max(0, output_tokens - 1)), - "egressNodeId": node_id, "errorCode": None, + "accountId": "101", "egressNodeId": node_id, "errorCode": None, }