fix(scheduler): admit credentials at their group frontier

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
+}

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)
+	}
+}