feat(flowcontrol): expose round-robin fairness identity via DumpState

pkg/epp/framework/plugins/flowcontrol/fairness/roundrobin/roundrobin.go

@@ -46,6 +46,8 @@ type roundRobin struct {
 	name string
 }
 
+var _ fwkplugin.StateDumper = &roundRobin{}
+
 func newRoundRobin(name string) *roundRobin {
 	if name == "" {
 		name = RoundRobinFairnessPolicyType
@@ -126,3 +128,27 @@ func (p *roundRobin) Pick(
 	c.lastSelected = nil
 	return nil, nil //nolint:nilnil
 }
+
+// roundRobinState is the debug snapshot for the round-robin fairness policy.
+// The policy is a stateless singleton: its per-priority-band cursors live in the
+// flow-control registry (PriorityBandAccessor.PolicyState), not on the plugin, so
+// no live scheduling state is reachable from here. This snapshot reports only
+// static identity plus an explicit marker so operators are not misled into
+// thinking the cursor position is shown.
+type roundRobinState struct {
+	Policy   string `json:"policy"`
+	Stateful bool   `json:"stateful"`
+	Note     string `json:"note"`
+}
+
+// DumpState implements [fwkplugin.StateDumper]. The round-robin policy holds no
+// mutable runtime state on the plugin; the per-band cursors are owned by the flow
+// registry. The payload is therefore a fixed, bounded struct independent of band
+// or flow cardinality.
+func (p *roundRobin) DumpState() (json.RawMessage, error) {
+	return json.Marshal(roundRobinState{
+		Policy:   RoundRobinFairnessPolicyType,
+		Stateful: false,
+		Note:     "round-robin cursors are tracked per priority band in the flow-control registry, not on this plugin",
+	})
+}

pkg/epp/framework/plugins/flowcontrol/fairness/roundrobin/roundrobin_test.go

@@ -18,6 +18,7 @@ package roundrobin
 
 import (
 	"context"
+	"encoding/json"
 	"fmt"
 	"sync"
 	"sync/atomic"
@@ -282,3 +283,88 @@ func TestRoundRobin_Pick_Concurrency(t *testing.T) {
 		})
 	}
 }
+
+func TestRoundRobin_DumpState(t *testing.T) {
+	t.Parallel()
+
+	// decode unmarshals the payload into the snapshot shape plus a catch-all so a
+	// stray extra field (e.g. a leaked flow ID or name) would be detectable.
+	decode := func(t *testing.T, payload json.RawMessage) (roundRobinState, map[string]json.RawMessage) {
+		t.Helper()
+		require.True(t, json.Valid(payload), "DumpState must return valid JSON")
+		var s roundRobinState
+		require.NoError(t, json.Unmarshal(payload, &s))
+		var raw map[string]json.RawMessage
+		require.NoError(t, json.Unmarshal(payload, &raw))
+		return s, raw
+	}
+
+	assertHonest := func(t *testing.T, s roundRobinState, raw map[string]json.RawMessage) {
+		t.Helper()
+		assert.Equal(t, RoundRobinFairnessPolicyType, s.Policy)
+		assert.False(t, s.Stateful, "policy holds no reachable mutable state")
+		assert.NotEmpty(t, s.Note, "note must explain where cursors actually live")
+		// Exactly the three sanitized fields, nothing else (no name, no flow IDs).
+		wantKeys := map[string]bool{"policy": true, "stateful": true, "note": true}
+		for k := range raw {
+			assert.Truef(t, wantKeys[k], "unexpected key %q in DumpState payload", k)
+		}
+		assert.Len(t, raw, len(wantKeys), "payload must contain only policy, stateful, note")
+	}
+
+	tests := []struct {
+		name   string
+		policy *roundRobin
+	}{
+		{name: "default name", policy: newRoundRobin("")},
+		{name: "custom name", policy: newRoundRobin("rr-a")},
+	}
+
+	var defaultBytes json.RawMessage
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			payload, err := tt.policy.DumpState()
+			require.NoError(t, err)
+			s, raw := decode(t, payload)
+			assertHonest(t, s, raw)
+			if tt.name == "default name" {
+				defaultBytes = payload
+			} else {
+				// Name is not emitted: a differently-named policy yields identical bytes.
+				assert.JSONEq(t, string(defaultBytes), string(payload),
+					"plugin name must not appear in the dump")
+			}
+		})
+	}
+
+	// Plugin-specific edge case: the dump cannot reflect cursor movement because
+	// the cursor lives in the band's PolicyState, not on the plugin. Advance a
+	// cursor via Pick, then assert DumpState is byte-identical to the untouched case.
+	t.Run("invariant after Pick", func(t *testing.T) {
+		policy := newRoundRobin("")
+		ctx := context.Background()
+		state := policy.NewState(ctx)
+		queue1 := &fwkfcmocks.MockFlowQueueAccessor{LenV: 1, FlowKeyV: flow1Key}
+		queue2 := &fwkfcmocks.MockFlowQueueAccessor{LenV: 2, FlowKeyV: flow2Key}
+		mockBand := &fwkfcmocks.MockPriorityBandAccessor{
+			PolicyStateV: state,
+			FlowKeysFunc: func() []flowcontrol.FlowKey { return []flowcontrol.FlowKey{flow1Key, flow2Key} },
+			QueueFunc: func(id string) flowcontrol.FlowQueueAccessor {
+				if id == flow1ID {
+					return queue1
+				}
+				return queue2
+			},
+		}
+		selected, err := policy.Pick(ctx, mockBand)
+		require.NoError(t, err)
+		require.NotNil(t, selected, "Pick should advance the band cursor")
+
+		payload, err := policy.DumpState()
+		require.NoError(t, err)
+		s, raw := decode(t, payload)
+		assertHonest(t, s, raw)
+		assert.JSONEq(t, string(defaultBytes), string(payload),
+			"DumpState must not and cannot reflect cursor movement")
+	})
+}