diff --git a/cmd/epp/runner/runner.go b/cmd/epp/runner/runner.go index c2fb54efbe..b106fe8a4c 100644 --- a/cmd/epp/runner/runner.go +++ b/cmd/epp/runner/runner.go @@ -743,7 +743,7 @@ func (r *Runner) parseConfigurationPhaseTwo(ctx context.Context, rawConfig *conf } // The plugins will be executed in topologically sorted order to ensure that data is produced before it is consumed. - r.requestControlConfig.OrderDataProducerPlugins(dag) + r.requestControlConfig.OrderPlugins(dag) r.parserRegistry = cfg.ParserRegistry logger.Info("loaded configuration from file/text successfully") diff --git a/pkg/epp/framework/interface/plugin/plugins.go b/pkg/epp/framework/interface/plugin/plugins.go index 25e76d3316..0f99290bdc 100644 --- a/pkg/epp/framework/interface/plugin/plugins.go +++ b/pkg/epp/framework/interface/plugin/plugins.go @@ -28,6 +28,14 @@ type Plugin interface { // DataDependencies holds the data keys a plugin consumes, split by whether they // are required (framework errors if no producer exists) or optional (framework // logs a warning but continues if no producer exists). +// +// The declared dependencies give every requestcontrol extension point a +// deterministic order: a plugin runs after the plugins producing the keys it +// consumes. Producer-before-consumer is the intended semantic on RequestHeader, +// Admit, Produce and PreRequest. ResponseHeader and ResponseBody are ordered the +// same way for determinism, but the correct direction there is unsettled - a +// producer may release consumed state in a response hook - so response-hook +// plugins must not rely on it. type DataDependencies struct { // Required keys — the framework will error at init time if no producer exists for any of these. Required map[DataKey]any diff --git a/pkg/epp/requestcontrol/request_control_config.go b/pkg/epp/requestcontrol/request_control_config.go index b26cdb4cf8..5a66cf8a2e 100644 --- a/pkg/epp/requestcontrol/request_control_config.go +++ b/pkg/epp/requestcontrol/request_control_config.go @@ -17,6 +17,8 @@ limitations under the License. package requestcontrol import ( + "sort" + "github.com/llm-d/llm-d-router/pkg/epp/framework/interface/plugin" fwkrc "github.com/llm-d/llm-d-router/pkg/epp/framework/interface/requestcontrol" ) @@ -108,17 +110,45 @@ func (c *Config) AddPlugins(pluginObjects ...plugin.Plugin) { } } -// OrderDataProducerPlugins reorders the DataProducer plugins in the Config based on the given sorted plugin names. -func (c *Config) OrderDataProducerPlugins(sortedPluginNames []string) { - sortedPlugins := make([]fwkrc.DataProducer, 0, len(sortedPluginNames)) - nameToPlugin := make(map[string]fwkrc.DataProducer) - for _, plugin := range c.dataProducerPlugins { - nameToPlugin[plugin.TypedName().String()] = plugin +// OrderPlugins reorders every extension point in the Config to the given sorted +// plugin names, so that a plugin runs after the plugins whose data it consumes. +// The names come from the data-dependency DAG, which only ranks producers and +// consumers; plugins it does not rank keep running, ordered by name after the +// ranked ones. +func (c *Config) OrderPlugins(sortedPluginNames []string) { + rank := make(map[string]int, len(sortedPluginNames)) + for i, name := range sortedPluginNames { + rank[name] = i } - for _, name := range sortedPluginNames { - if plugin, ok := nameToPlugin[name]; ok { - sortedPlugins = append(sortedPlugins, plugin) + + c.requestHeaderPlugins = orderByName(c.requestHeaderPlugins, rank) + c.admissionPlugins = orderByName(c.admissionPlugins, rank) + c.dataProducerPlugins = orderByName(c.dataProducerPlugins, rank) + c.preRequestPlugins = orderByName(c.preRequestPlugins, rank) + c.responseReceivedPlugins = orderByName(c.responseReceivedPlugins, rank) + c.responseStreamingPlugins = orderByName(c.responseStreamingPlugins, rank) +} + +// orderByName returns the plugins ordered by their rank, followed by the unranked +// ones sorted by name. Sorting the unranked plugins rather than keeping their +// original order matters: they reach the Config through a map iteration, so their +// incoming order differs between runs. +func orderByName[T plugin.Plugin](plugins []T, rank map[string]int) []T { + ordered := make([]T, len(plugins)) + copy(ordered, plugins) + + sort.SliceStable(ordered, func(i, j int) bool { + iName, jName := ordered[i].TypedName().String(), ordered[j].TypedName().String() + iRank, iRanked := rank[iName] + jRank, jRanked := rank[jName] + if iRanked != jRanked { + return iRanked } - } - c.dataProducerPlugins = sortedPlugins + if !iRanked { + return iName < jName + } + return iRank < jRank + }) + + return ordered } diff --git a/pkg/epp/requestcontrol/request_control_config_test.go b/pkg/epp/requestcontrol/request_control_config_test.go new file mode 100644 index 0000000000..732665803c --- /dev/null +++ b/pkg/epp/requestcontrol/request_control_config_test.go @@ -0,0 +1,264 @@ +/* +Copyright 2026 llm-d Authors. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +package requestcontrol + +import ( + "context" + "fmt" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "github.com/llm-d/llm-d-router/pkg/epp/datalayer" + fwkdl "github.com/llm-d/llm-d-router/pkg/epp/framework/interface/datalayer" + fwkplugin "github.com/llm-d/llm-d-router/pkg/epp/framework/interface/plugin" + fwkrc "github.com/llm-d/llm-d-router/pkg/epp/framework/interface/requestcontrol" + fwksched "github.com/llm-d/llm-d-router/pkg/epp/framework/interface/scheduling" +) + +// orderTestData is the payload type exchanged between the producer and consumer +// mocks below. buildDAG compares produced and consumed types, so both sides must +// declare the same one. +type orderTestData struct{} + +func (d *orderTestData) Clone() fwkdl.Cloneable { return &orderTestData{} } + +// hookPlugin implements every requestcontrol extension point plus Produces and +// Consumes, so a single instance lands on all six Config hook lists and is ranked +// by the data-dependency DAG. Leaving produces or consumes nil makes the instance +// a pure consumer or a pure producer respectively. +type hookPlugin struct { + name string + produces map[fwkplugin.DataKey]any + consumes map[fwkplugin.DataKey]any +} + +func (p *hookPlugin) TypedName() fwkplugin.TypedName { + return fwkplugin.TypedName{Name: p.name, Type: "mock"} +} +func (p *hookPlugin) Produces() map[fwkplugin.DataKey]any { return p.produces } +func (p *hookPlugin) Consumes() fwkplugin.DataDependencies { + return fwkplugin.DataDependencies{Required: p.consumes} +} +func (p *hookPlugin) Produce(_ context.Context, _ *fwksched.InferenceRequest, _ []fwksched.Endpoint) error { + return nil +} +func (p *hookPlugin) PreRequest(_ context.Context, _ *fwksched.InferenceRequest, _ *fwksched.SchedulingResult) { +} +func (p *hookPlugin) RequestHeader(_ context.Context, _ *fwksched.InferenceRequest) error { return nil } +func (p *hookPlugin) ResponseHeader(_ context.Context, _ *fwksched.InferenceRequest, _ *fwkrc.Response, _ *fwkdl.EndpointMetadata) { +} +func (p *hookPlugin) ResponseBody(_ context.Context, _ *fwksched.InferenceRequest, _ *fwkrc.Response, _ *fwkdl.EndpointMetadata) { +} +func (p *hookPlugin) Admit(_ context.Context, _ *fwksched.InferenceRequest, _ []fwksched.Endpoint) error { + return nil +} + +// unrankedPlugin implements a hook but is neither a producer nor a consumer, so +// the data-dependency DAG never ranks it. Ordering must not drop it. +type unrankedPlugin struct { + name string +} + +func (p *unrankedPlugin) TypedName() fwkplugin.TypedName { + return fwkplugin.TypedName{Name: p.name, Type: "mock"} +} +func (p *unrankedPlugin) PreRequest(_ context.Context, _ *fwksched.InferenceRequest, _ *fwksched.SchedulingResult) { +} + +func names[T interface{ TypedName() fwkplugin.TypedName }](plugins []T) []string { + result := make([]string, 0, len(plugins)) + for _, p := range plugins { + result = append(result, p.TypedName().String()) + } + return result +} + +// TestConfig_OrderPlugins_AllHookLists drives the ordering path the runner uses +// (parseConfigurationPhaseTwo: AddPlugins over the handle's plugins, then apply the +// data-dependency order) and asserts every extension point honours the DAG, not +// just Produce. +func TestConfig_OrderPlugins_AllHookLists(t *testing.T) { + key := fwkplugin.NewDataKey("orderTestData", "mock") + producer := &hookPlugin{name: "P", produces: map[fwkplugin.DataKey]any{key: &orderTestData{}}} + consumer := &hookPlugin{name: "C", consumes: map[fwkplugin.DataKey]any{key: &orderTestData{}}} + unranked := &unrankedPlugin{name: "U"} + + cfg := NewConfig() + // Deliberately consumer-first: this is the order GetAllPlugins may hand over, + // since it ranges a map. + cfg.AddPlugins(consumer, unranked, producer) + + sorted, err := datalayer.ValidateAndOrderDataDependencies([]fwkplugin.Plugin{consumer, unranked, producer}) + require.NoError(t, err) + require.Equal(t, []string{"P/mock", "C/mock"}, sorted, + "DAG must rank the producer before the consumer and must not rank the unranked plugin") + + cfg.OrderPlugins(sorted) + + testCases := []struct { + name string + got []string + want []string + }{ + { + // The regression from #1856: a consumer's PreRequest ran before the + // producer's, depending on map order. + name: "preRequest orders producer before consumer and keeps unranked plugins", + got: names(cfg.preRequestPlugins), + want: []string{"P/mock", "C/mock", "U/mock"}, + }, + { + name: "requestHeader orders producer before consumer", + got: names(cfg.requestHeaderPlugins), + want: []string{"P/mock", "C/mock"}, + }, + { + name: "responseReceived orders producer before consumer", + got: names(cfg.responseReceivedPlugins), + want: []string{"P/mock", "C/mock"}, + }, + { + name: "responseStreaming orders producer before consumer", + got: names(cfg.responseStreamingPlugins), + want: []string{"P/mock", "C/mock"}, + }, + { + name: "admission orders producer before consumer", + got: names(cfg.admissionPlugins), + want: []string{"P/mock", "C/mock"}, + }, + { + // Already correct before this change; guards against regressing it. + name: "dataProducer orders producer before consumer", + got: names(cfg.dataProducerPlugins), + want: []string{"P/mock", "C/mock"}, + }, + } + + for _, tc := range testCases { + t.Run(tc.name, func(t *testing.T) { + assert.Equal(t, tc.want, tc.got) + }) + } +} + +// TestConfig_OrderPlugins_EdgeCases covers the boundaries the DAG path can hand to +// OrderPlugins: nothing configured, nothing ranked, and names for plugins that are +// not on a given list. +func TestConfig_OrderPlugins_EdgeCases(t *testing.T) { + key := fwkplugin.NewDataKey("orderTestData", "mock") + + testCases := []struct { + name string + plugins []fwkplugin.Plugin + sortedNames []string + wantPreReq []string + }{ + { + name: "empty config with empty order", + plugins: nil, + sortedNames: nil, + wantPreReq: []string{}, + }, + { + name: "no plugin is ranked: unranked plugins are kept, sorted by name", + plugins: []fwkplugin.Plugin{&unrankedPlugin{name: "U2"}, &unrankedPlugin{name: "U1"}}, + sortedNames: nil, + wantPreReq: []string{"U1/mock", "U2/mock"}, + }, + { + name: "order contains names absent from the config", + plugins: []fwkplugin.Plugin{&unrankedPlugin{name: "U1"}}, + sortedNames: []string{"ghost/mock", "U1/mock"}, + wantPreReq: []string{"U1/mock"}, + }, + { + name: "ranked and unranked mixed: ranked first, unranked appended by name", + plugins: []fwkplugin.Plugin{ + &unrankedPlugin{name: "U2"}, + &hookPlugin{name: "C", consumes: map[fwkplugin.DataKey]any{key: &orderTestData{}}}, + &unrankedPlugin{name: "U1"}, + &hookPlugin{name: "P", produces: map[fwkplugin.DataKey]any{key: &orderTestData{}}}, + }, + sortedNames: []string{"P/mock", "C/mock"}, + wantPreReq: []string{"P/mock", "C/mock", "U1/mock", "U2/mock"}, + }, + } + + for _, tc := range testCases { + t.Run(tc.name, func(t *testing.T) { + cfg := NewConfig() + cfg.AddPlugins(tc.plugins...) + cfg.OrderPlugins(tc.sortedNames) + assert.Equal(t, tc.wantPreReq, names(cfg.preRequestPlugins)) + }) + } +} + +// TestConfig_OrderPlugins_InputOrderIndependent is the core #1856/#2040 invariant: +// AddPlugins receives plugins in Go's randomized map-iteration order (from +// GetAllPlugins), so the ordered hook lists must be a pure function of the DAG and +// the plugin names, not of the order plugins were handed over. Each fixed permutation +// below stands in for one possible map-iteration order and must produce the same +// result. +func TestConfig_OrderPlugins_InputOrderIndependent(t *testing.T) { + key := fwkplugin.NewDataKey("orderTestData", "mock") + // Base set indexed for permutation: 0=P (producer), 1=C (consumer), 2=U1, 3=U2. + newSet := func() []fwkplugin.Plugin { + return []fwkplugin.Plugin{ + &hookPlugin{name: "P", produces: map[fwkplugin.DataKey]any{key: &orderTestData{}}}, + &hookPlugin{name: "C", consumes: map[fwkplugin.DataKey]any{key: &orderTestData{}}}, + &unrankedPlugin{name: "U1"}, + &unrankedPlugin{name: "U2"}, + } + } + // Producer before consumer (DAG); unranked appended by name. Identical for every + // input order. + wantPreReq := []string{"P/mock", "C/mock", "U1/mock", "U2/mock"} + wantRanked := []string{"P/mock", "C/mock"} // lists the unranked plugins do not join + + permutations := [][]int{ + {0, 1, 2, 3}, + {3, 2, 1, 0}, + {1, 3, 0, 2}, + {2, 0, 3, 1}, + } + for _, perm := range permutations { + t.Run(fmt.Sprintf("input order %v", perm), func(t *testing.T) { + base := newSet() + shuffled := make([]fwkplugin.Plugin, len(perm)) + for i, idx := range perm { + shuffled[i] = base[idx] + } + + cfg := NewConfig() + cfg.AddPlugins(shuffled...) + sorted, err := datalayer.ValidateAndOrderDataDependencies(shuffled) + require.NoError(t, err) + cfg.OrderPlugins(sorted) + + assert.Equal(t, wantPreReq, names(cfg.preRequestPlugins), "preRequest") + assert.Equal(t, wantRanked, names(cfg.requestHeaderPlugins), "requestHeader") + assert.Equal(t, wantRanked, names(cfg.admissionPlugins), "admission") + assert.Equal(t, wantRanked, names(cfg.responseReceivedPlugins), "responseReceived") + assert.Equal(t, wantRanked, names(cfg.responseStreamingPlugins), "responseStreaming") + assert.Equal(t, wantRanked, names(cfg.dataProducerPlugins), "dataProducer") + }) + } +} diff --git a/pkg/epp/util/utils.go b/pkg/epp/util/utils.go index ca936e9303..418783644f 100644 --- a/pkg/epp/util/utils.go +++ b/pkg/epp/util/utils.go @@ -23,6 +23,11 @@ import ( // TopologicalSort performs Kahn's Algorithm on a DAG. // It returns the sorted order or an error if a cycle is detected. +// +// The order is a pure function of the graph: nodes that are ready at the same +// time are dequeued largest name first, which the final reverse turns into +// ascending name order. Without that tie-break the result would follow Go's +// randomized map iteration and change from one call to the next. func TopologicalSort(graph map[string][]string) ([]string, error) { // 1. Initialize in-degree map inDegree := make(map[string]int) @@ -49,9 +54,11 @@ func TopologicalSort(graph map[string][]string) ([]string, error) { // 3. Process the queue for len(queue) > 0 { - // Dequeue - u := queue[0] - queue = queue[1:] + // Dequeue the largest ready node. Kahn's Algorithm accepts any ready + // node, so the choice is free to serve as the tie-break. + slices.Sort(queue) + u := queue[len(queue)-1] + queue = queue[:len(queue)-1] result = append(result, u) diff --git a/pkg/epp/util/utils_test.go b/pkg/epp/util/utils_test.go new file mode 100644 index 0000000000..96664653a4 --- /dev/null +++ b/pkg/epp/util/utils_test.go @@ -0,0 +1,114 @@ +/* +Copyright 2026 llm-d Authors. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +package util + +import ( + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// runs is high enough that Go's randomized map iteration would surface a +// map-order dependency in the sort. +const runs = 100 + +// TestTopologicalSort_Deterministic pins the ordering contract: the result is a +// pure function of the graph, dependencies precede dependents, and nodes with no +// dependency relation come out in ascending name order. +func TestTopologicalSort_Deterministic(t *testing.T) { + testCases := []struct { + name string + graph map[string][]string + want []string + wantErr string + }{ + { + name: "empty graph", + graph: map[string][]string{}, + }, + { + name: "single node", + graph: map[string][]string{"A/mock": {}}, + want: []string{"A/mock"}, + }, + { + name: "independent nodes come out in ascending name order", + graph: map[string][]string{"B/mock": {}, "A/mock": {}, "C/mock": {}}, + want: []string{"A/mock", "B/mock", "C/mock"}, + }, + { + // C depends on A; B and D depend on nothing. + name: "dependency respected, independent nodes ascending", + graph: map[string][]string{ + "C/mock": {"A/mock"}, + "A/mock": {}, + "B/mock": {}, + "D/mock": {}, + }, + want: []string{"A/mock", "B/mock", "C/mock", "D/mock"}, + }, + { + name: "linear chain", + graph: map[string][]string{ + "C/mock": {"B/mock"}, + "B/mock": {"A/mock"}, + "A/mock": {}, + }, + want: []string{"A/mock", "B/mock", "C/mock"}, + }, + { + // D depends on B and C, both of which depend on A. + name: "diamond", + graph: map[string][]string{ + "D/mock": {"B/mock", "C/mock"}, + "B/mock": {"A/mock"}, + "C/mock": {"A/mock"}, + "A/mock": {}, + }, + want: []string{"A/mock", "B/mock", "C/mock", "D/mock"}, + }, + { + name: "topological order wins over name order", + graph: map[string][]string{ + "A/mock": {"Z/mock"}, + "Z/mock": {}, + }, + want: []string{"Z/mock", "A/mock"}, + }, + { + name: "cycle is rejected", + graph: map[string][]string{"X/mock": {"Y/mock"}, "Y/mock": {"X/mock"}}, + wantErr: "cycle detected", + }, + } + + for _, tc := range testCases { + t.Run(tc.name, func(t *testing.T) { + for i := range runs { + got, err := TopologicalSort(tc.graph) + if tc.wantErr != "" { + require.Error(t, err) + assert.Contains(t, err.Error(), tc.wantErr) + continue + } + require.NoError(t, err) + assert.Equal(t, tc.want, got, "run %d returned a different order", i) + } + }) + } +}