From 47dd4a38c3877d0575e8c1a88c0f87e6c727167d Mon Sep 17 00:00:00 2001 From: tbphp Date: Sat, 7 Nov 2026 13:00:22 +0800 Subject: [PATCH] fix(scheduler): admit credentials at their group frontier Credentials that join or rejoin rotation while the service is running were admitted at the highest progress of the whole candidate pool, which can span groups, or at the global watermark. In a group-only model they then waited hundreds of rounds behind their own peers and received no traffic. Admission now aligns with the leading progress of the member's own group across the ledger, so it also holds when active peers are absent from the current pool (model cooldown, tried retries, credential filters). Groups whose members are all joining keep the previous pool baseline. Closes #945 --- internal/scheduler/fair.go | 26 ++++++++- internal/scheduler/fairness_test.go | 86 +++++++++++++++++++++++++++++ 2 files changed, 111 insertions(+), 1 deletion(-) diff --git a/internal/scheduler/fair.go b/internal/scheduler/fair.go index 4d072e3f07..ef0bb36404 100644 --- a/internal/scheduler/fair.go +++ b/internal/scheduler/fair.go @@ -60,12 +60,22 @@ func (iterator *Iterator) selectCredential(candidates []weightedCredential, pref return } // 同批成员共用已有成员的领先进度，不能继承落后成员的历史欠额。 + // 入场对齐本组在轮转成员的最前进度，不受其他分组或本次缺席同组成员影响； + // 整组都待入场时才沿用候选池基线。 + var frontiers map[uint]state.SchedulingProgress for _, candidate := range eligible { member := ledger.Members[candidate.meta.ID] // 旧请求可以继续记账，但不能用过期候选池完成新配置的入场校准。 if member.Pending && (!ledger.GroupsKnown || ledger.Groups[member.GroupID]) && iterator.snapshot != nil && iterator.snapshot.Revision >= ledger.GroupRevision { - member.Admit(baseline) + if frontiers == nil { + frontiers = groupFrontiers(ledger) + } + if frontier, ok := frontiers[member.GroupID]; ok { + member.Admit(frontier) + } else { + member.Admit(baseline) + } } } less := func(left, right weightedCredential) bool { @@ -125,3 +135,17 @@ func (iterator *Iterator) selectCredential(candidates []weightedCredential, pref }) return selected, found } + +// groupFrontiers 只在有成员入场时扫描账本，返回各分组在轮转成员的最前进度。 +func groupFrontiers(ledger *state.SchedulingLedger) map[uint]state.SchedulingProgress { + frontiers := make(map[uint]state.SchedulingProgress) + for _, member := range ledger.Members { + if member.Pending { + continue + } + if current, ok := frontiers[member.GroupID]; !ok || member.Progress.Compare(current) > 0 { + frontiers[member.GroupID] = member.Progress + } + } + return frontiers +} diff --git a/internal/scheduler/fairness_test.go b/internal/scheduler/fairness_test.go index 18a4b196d1..dc19611226 100644 --- a/internal/scheduler/fairness_test.go +++ b/internal/scheduler/fairness_test.go @@ -390,3 +390,89 @@ func TestFairnessChargesRetriesButKeepsTriedRequestLocal(t *testing.T) { } }) } + +// 分组 1、2 共享 shared，各有独占模型；分组 3 只有独占模型，用于推高全局水位。 +func admissionFixture(t *testing.T, credentials ...state.CredentialEntry) (*state.ConfigSnapshot, *state.CredentialRegistry) { + t.Helper() + snapshot, err := state.Compile(state.CompileInput{ChannelRegistry: channel.NewRegistry(), Groups: []state.GroupConfig{ + {ID: 1, Name: "one", ChannelID: channel.OpenAI, ConnectionType: "api_key", Enabled: true, + Models: []state.ModelConfig{{ID: "shared"}, {ID: "only-one"}}}, + {ID: 2, Name: "two", ChannelID: channel.OpenAI, ConnectionType: "api_key", Enabled: true, + Models: []state.ModelConfig{{ID: "shared"}, {ID: "only-two"}}}, + {ID: 3, Name: "three", ChannelID: channel.OpenAI, ConnectionType: "api_key", Enabled: true, + Models: []state.ModelConfig{{ID: "only-three"}}}, + }}) + if err != nil { + t.Fatal(err) + } + registry := state.NewCredentialRegistry() + if err := registry.ReplaceCredentials(credentials); err != nil { + t.Fatal(err) + } + return snapshot, registry +} + +func admissionEntry(id, groupID uint) state.CredentialEntry { + return state.CredentialEntry{ID: id, GroupID: groupID, Version: 1, IdentityGeneration: 1, + Status: state.CredentialStatusActive, Fingerprint: "fixture", EncryptedValue: "cipher"} +} + +func admissionPick(t *testing.T, snapshot *state.ConfigSnapshot, registry *state.CredentialRegistry, model string) uint { + t.Helper() + selection, err := New(snapshot, registry, Query{ClientProtocol: protocol.OpenAICompletions, + Operation: execution.OperationChatCompletion, ExternalModel: modelPointer(model)}).Next() + if err != nil { + t.Fatal(err) + } + return selection.CredentialID +} + +func admissionDistribution(t *testing.T, snapshot *state.ConfigSnapshot, registry *state.CredentialRegistry, model string, requests int) map[uint]int { + t.Helper() + counts := make(map[uint]int) + for range requests { + counts[admissionPick(t, snapshot, registry, model)]++ + } + return counts +} + +func TestFairnessAdmissionIgnoresOtherGroupsProgress(t *testing.T) { + snapshot, registry := admissionFixture(t, admissionEntry(11, 1), admissionEntry(12, 1), admissionEntry(21, 2)) + admissionDistribution(t, snapshot, registry, "only-two", 1000) + admissionDistribution(t, snapshot, registry, "only-one", 100) + if err := registry.ApplyCredentialImport(1, []state.CredentialEntry{admissionEntry(13, 1)}); err != nil { + t.Fatal(err) + } + // 新凭据首次出现在跨分组候选池，不能继承分组 2 更靠前的进度。 + admissionPick(t, snapshot, registry, "shared") + counts := admissionDistribution(t, snapshot, registry, "only-one", 300) + if counts[11] != 100 || counts[12] != 100 || counts[13] != 100 { + t.Fatalf("group-only distribution = %v, want 100 each", counts) + } +} + +func TestFairnessAdmissionUsesGroupFrontierWhenPeersAreAbsent(t *testing.T) { + snapshot, registry := admissionFixture(t, admissionEntry(11, 1), admissionEntry(21, 2), admissionEntry(31, 3)) + admissionDistribution(t, snapshot, registry, "only-three", 1000) + admissionDistribution(t, snapshot, registry, "only-two", 500) + admissionDistribution(t, snapshot, registry, "only-one", 100) + // 11 只对 shared 冷却，仍在分组 1 轮转，但不在下一次 shared 候选池里。 + now := time.Now() + ref, ok := registry.CredentialRef(11) + if !ok { + t.Fatal("credential 11 missing") + } + if accepted, _ := registry.SetModelCooldown(ref, "shared", now.Add(time.Hour), now); !accepted { + t.Fatal("model cooldown rejected") + } + if err := registry.ApplyCredentialImport(1, []state.CredentialEntry{admissionEntry(13, 1)}); err != nil { + t.Fatal(err) + } + if got := admissionPick(t, snapshot, registry, "shared"); got != 13 && got != 21 { + t.Fatalf("shared selected %d, want a credential outside the cooldown", got) + } + counts := admissionDistribution(t, snapshot, registry, "only-one", 200) + if counts[11] != 100 || counts[13] != 100 { + t.Fatalf("group-only distribution = %v, want 100 each", counts) + } +}