diff --git a/docs/subsystems/gateway-routing.md b/docs/subsystems/gateway-routing.md index 18fc4b12..2922c57c 100644 --- a/docs/subsystems/gateway-routing.md +++ b/docs/subsystems/gateway-routing.md @@ -42,13 +42,14 @@ served by a plugin are in [Provider and plugin ownership](provider-plugins.md). - A fallback happens only while none of the reply has been sent. An agent never gets half a reply from one upstream and the rest from another. - A resting candidate is tried last, never dropped (`restLast`). A lone candidate is tried even while it rests, because there is no other. - Routing decides who goes first only when nobody has answered the conversation yet. Affinity keeps the answerer while its cache is worth it. Auto mode keeps it across turns while at least `cacheWorth` (1024) tokens were read from the cache and the cache isn't cold (`cacheCold`, 5 minutes). +- In turn (`rotate`), a conversation nobody has answered yet goes to the key, account or model the fewest other conversations of the provider's model or the group answered in the last `heldFor` (30 minutes) are on (`leastHeld`, #946). Ties, resting ones and a group with a group in it keep routing's order. With session affinity, two sessions of one group so stay on different models. - Different failures are classified separately: out of credit, out of quota, rate limit and other errors rest for different lengths. A fix must not merge them into one bucket (see #873). - Group members are `provider/model[:effort]`. A member with a fixed effort is asked for that effort, bounded by the group's levels (#671). ## Verification ```sh -go test -tags nogui ./internal/gateway -run 'TestRouting|TestSmartRouting|TestFallback|TestNoFallbackForOtherErrors|TestLastFallbackErrorReachesTheAgent|TestRateLimitedSinksToTheBack|TestGroupRateLimitedSinks|TestAffinity|TestSeveralKeysOnTakeOverFromEachOther|TestSubscriptionAccountsTakeOver|TestGroup|TestTrace' +go test -tags nogui ./internal/gateway -run 'TestRotateSpreadsSessions|TestRouting|TestSmartRouting|TestFallback|TestNoFallbackForOtherErrors|TestLastFallbackErrorReachesTheAgent|TestRateLimitedSinksToTheBack|TestGroupRateLimitedSinks|TestAffinity|TestSeveralKeysOnTakeOverFromEachOther|TestSubscriptionAccountsTakeOver|TestGroup|TestTrace' go test -tags nogui ./internal/provider -run 'TestGroup' ``` diff --git a/internal/gateway/affinity.go b/internal/gateway/affinity.go index 37e9dc0d..358f2450 100644 --- a/internal/gateway/affinity.go +++ b/internal/gateway/affinity.go @@ -27,6 +27,7 @@ import ( "path/filepath" "regexp" "slices" + "strings" "sync" "time" @@ -293,10 +294,71 @@ func affine(scope, mode string, rotate bool, in http.Header, from provider.Proto if rotate { cs, pl = after(cs, pl, at) } + case "first": + if rotate { + cs, pl = leastHeld(scope, cs, pl) + } } return cs, pl, a, key } +// heldFor is how long after its last answer a conversation still counts as +// holding the one that answered it: a session whose user is reading or +// typing is still going. +const heldFor = 30 * time.Minute + +// leastHeld puts first, for a conversation nobody has answered yet, the +// one in turn that the fewest other conversations in scope are held on +// (#946): in turn counts requests, not conversations, so a second session +// started after the first had sent an even number of them went to the +// model the first was kept on. Of those held on as few, routing's order +// stands. +func leastHeld(scope string, cs []candidate, pl planned) ([]candidate, planned) { + if len(cs) < 2 || len(pl.order) < len(cs) { + return cs, pl + } + if strings.HasPrefix(scope, provider.GroupPrefix) { + if _, ms, ok := provider.FindGroup(scope); ok { + for _, m := range ms { + if len(m.Path) > 1 { + return cs, pl // a group in the group: its own routing orders its models (#576) + } + } + } + } + now := time.Now() + held := map[string]int{} + sticks.Lock() + for k, st := range sticks.m { + if strings.HasPrefix(k, scope+"|") && now.Sub(st.at) <= heldFor { + held[st.who+"/"+provider.WithMemberEffort(st.model, st.effort)]++ + } + } + sticks.Unlock() + if len(held) == 0 { + return cs, pl + } + seat := func(c candidate) string { return c.who() + "/" + provider.WithMemberEffort(c.model, c.effort) } + best := -1 + for i, c := range cs { + if pl.order[i].Rest != nil || pl.order[i].Aside { + continue + } + if best < 0 || held[seat(c)] < held[seat(cs[best])] { + best = i + } + } + if best <= 0 { + return cs, pl + } + cs = append(append([]candidate{cs[best]}, cs[:best]...), cs[best+1:]...) + pl.order = append(append([]Weighed{pl.order[best]}, pl.order[:best]...), pl.order[best+1:]...) + for j := range pl.order { + pl.order[j].Turn = j == 0 + } + return cs, pl +} + // after puts first the one after cs[at], round from the end, that isn't // resting. func after(cs []candidate, pl planned, at int) ([]candidate, planned) { diff --git a/internal/gateway/group_test.go b/internal/gateway/group_test.go index 5e5681cb..ebd7cc58 100644 --- a/internal/gateway/group_test.go +++ b/internal/gateway/group_test.go @@ -430,3 +430,64 @@ func (k *geminiKeyed) ServeHTTP(w http.ResponseWriter, r *http.Request) { `data: [DONE]`, )) } + +// A group in turn keeping each session where it began: a second session +// goes to the model the first isn't on, however many requests the first +// sent before it (#946); one resting is still passed over. +func TestRotateSpreadsSessions(t *testing.T) { + pair := func() (*keyed, *keyed, *Server) { + fresh(t) + a, b := &keyed{}, &keyed{} + serveOn(t, "ga", "ka", []string{"m1"}, a) + serveOn(t, "gb", "kb", []string{"m2"}, b) + if err := provider.SaveGroup(provider.Group{Name: "Pair", Members: []string{"ga/m1", "gb/m2"}, Routing: provider.Rotate, Affinity: provider.AffinitySession}); err != nil { + t.Fatal(err) + } + routed.Lock() + delete(routed.turn, provider.GroupPrefix+"pair") + routed.Unlock() + return a, b, New() + } + // which of the two the request went to + went := func(a, b *keyed, post func()) string { + na, nb := len(a.tried), len(b.tried) + post() + switch { + case len(a.tried) > na && len(b.tried) == nb: + return "m1" + case len(b.tried) > nb && len(a.tried) == na: + return "m2" + } + t.Fatalf("tried %v and %v", a.tried, b.tried) + return "" + } + for n := 1; n <= 4; n++ { + a, b, s := pair() + msgs := `{"role":"user","content":"hi"}` + var first string + for i := range n { + if i > 0 { + msgs += `,{"role":"assistant","content":"yo"},{"role":"user","content":"more"}` + } + first = went(a, b, func() { postAs(t, s, "s1", `{"model":"group/pair","messages":[`+msgs+`]}`) }) + } + second := went(a, b, func() { postAs(t, s, "s2", `{"model":"group/pair","messages":[{"role":"user","content":"other"}]}`) }) + if first == second { + t.Errorf("after %d requests of s1 on %s, s2 went there too", n, first) + } + if r := s.trace.routes[len(s.trace.routes)-1]; r.Affinity == nil || r.Affinity.Why != "first" { + t.Errorf("s2's affinity: %+v", r.Affinity) + } + } + + // the one fewest are on is resting: routing's order stands + a, b, s := pair() + first := went(a, b, func() { postAs(t, s, "s1", `{"model":"group/pair","messages":[{"role":"user","content":"hi"}]}`) }) + other := map[string]string{"m1": "gb", "m2": "ga"}[first] + restingUntil.Lock() + restingUntil.m[other] = time.Now().Add(time.Minute) + restingUntil.Unlock() + if got := went(a, b, func() { postAs(t, s, "s2", `{"model":"group/pair","messages":[{"role":"user","content":"other"}]}`) }); got != first { + t.Fatalf("s2 went to %s, where %s rests", got, other) + } +}