fix several minor issues and update comments

Author: RakhithaRR

gateway/gateway-controller/pkg/models/runtime_deploy_config.go

@@ -333,6 +333,17 @@ func (rdc *RuntimeDeployConfig) ValidateResolution() error {
 						"route %q: canonical chain key %q belongs to routing partition (vhost) %q, but the route serves %q",
 						routeKey, canonical, vhost, route.Vhost)
 				}
+			} else if canonical != routeKey {
+				// A composed operation key is the one redirect an identity route may
+				// carry. Any other key that merely happens to exist is refused, because
+				// the failure it hides is silent: a route pointed at another route's
+				// chain — a public route carrying "GET|/admin|h", say — passes the
+				// existence check above and then runs that route's authentication and
+				// rate limits instead of its own. Same class as the cross-partition case,
+				// without a composed key's structure to detect it from.
+				return fmt.Errorf(
+					"route %q: canonical chain key %q is neither the route key nor a composed operation key",
+					routeKey, canonical)
 			}
 			continue
 		}

gateway/gateway-controller/pkg/policyxds/route_resolution_test.go

@@ -151,7 +151,7 @@ func TestEmptyResolutionFieldsAreOmitted(t *testing.T) {
 	}
 }
 
-// The §5.2 golden: the complete emitted content for an existing kind's route, pinned
+// Golden test: the complete emitted content for an existing kind's route, pinned
 // value by value. It is a content comparison rather than a byte comparison on purpose
 // — the resource bytes are produced by anypb.New over a Struct whose map fields have
 // no defined wire order, and the LinearCache re-versions every resource it is handed
@@ -301,11 +301,36 @@ func TestValidateResolution(t *testing.T) {
 			},
 		},
 		{
-			name: "identity route with an explicit canonical key",
+			name: "identity route with an explicit canonical key equal to its route key",
+			rdc: &models.RuntimeDeployConfig{
+				Routes:       map[string]*models.Route{"GET|/pets|h": {CanonicalChainKey: "GET|/pets|h"}},
+				PolicyChains: chains("GET|/pets|h"),
+			},
+		},
+		{
+			// A chain that exists but is neither this route's key nor a composed operation
+			// key. Existence alone would accept it, and the route would then run whatever
+			// policies that chain carries — the borrowed-policies failure is silent, so it
+			// has to be refused here.
+			name: "identity route pointed at an arbitrary existing chain",
 			rdc: &models.RuntimeDeployConfig{
 				Routes:       map[string]*models.Route{"GET|/pets|h": {CanonicalChainKey: "shared-chain"}},
-				PolicyChains: chains("shared-chain"),
+				PolicyChains: chains("GET|/pets|h", "shared-chain"),
+			},
+			wantErr: `is neither the route key nor a composed operation key`,
+		},
+		{
+			// The concrete shape of that mistake: a public route carrying another route's
+			// key, which would silently borrow that route's authentication.
+			name: "identity route pointed at another route's chain",
+			rdc: &models.RuntimeDeployConfig{
+				Routes: map[string]*models.Route{
+					"GET|/pets|h":  {CanonicalChainKey: "GET|/admin|h"},
+					"GET|/admin|h": {},
+				},
+				PolicyChains: chains("GET|/pets|h", "GET|/admin|h"),
 			},
+			wantErr: `canonical chain key "GET|/admin|h" is neither the route key nor a composed operation key`,
 		},
 		{
 			// An identity route deliberately redirected to a composed operation key:

gateway/gateway-controller/pkg/policyxds/snapshot.go

@@ -417,8 +417,8 @@ func (t *Translator) createRouteConfigResource(
 		data["resolver_config"] = decoded
 	}
 	// Omitted when empty, like every other resolution field: a route that declares
-	// nothing must serialise byte-identically to how it did before the field existed
-	// (§5.2), or every existing kind's RouteConfig re-versions for no reason.
+	// nothing must serialise exactly as it did before the field existed, or every
+	// existing kind's RouteConfig re-versions for no reason.
 	if route.ResponseKind != "" {
 		data["response_kind"] = route.ResponseKind
 	}

gateway/gateway-runtime/policy-engine/cmd/policy-engine/main.go

@@ -263,7 +263,7 @@ func main() {
 		slog.InfoContext(ctx, "Policy Engine listening on TCP port", "port", cfg.PolicyEngine.Server.ExtProcPort)
 	}
 
-	grpcServer := grpc.NewServer()
+	grpcServer := grpc.NewServer(extProcServerOptions(cfg)...)
 	extprocv3.RegisterExternalProcessorServer(grpcServer, extprocServer)
 
 	// Enable block/mutex profiling sampling when pprof is enabled. These are the
@@ -452,3 +452,23 @@ func initializeFileConfig(ctx context.Context, cfg *config.Config, k *kernel.Ker
 
 	return nil
 }
+
+// extProcServerOptions bounds the ext_proc gRPC server explicitly, rather than taking
+// gRPC's defaults: the receive default is 4 MiB whatever the body ceilings are configured
+// to be, the send default is unbounded, and the concurrent-stream default is effectively
+// unlimited. This is the hottest gRPC server in the data plane, so all three are set from
+// validated configuration (see config.Config.Validate, which refuses to start when a
+// message limit is below what the body ceilings require).
+func extProcServerOptions(cfg *config.Config) []grpc.ServerOption {
+	server := cfg.PolicyEngine.Server
+	slog.Info("ext_proc gRPC server limits",
+		"max_recv_msg_bytes", server.MaxRecvMsgBytes,
+		"max_send_msg_bytes", server.MaxSendMsgBytes,
+		"max_concurrent_streams", server.MaxConcurrentStreams)
+
+	return []grpc.ServerOption{
+		grpc.MaxRecvMsgSize(int(server.MaxRecvMsgBytes)),
+		grpc.MaxSendMsgSize(int(server.MaxSendMsgBytes)),
+		grpc.MaxConcurrentStreams(server.MaxConcurrentStreams),
+	}
+}

gateway/gateway-runtime/policy-engine/cmd/policy-engine/main_test.go

@@ -28,6 +28,7 @@ import (
 
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
+	"google.golang.org/grpc"
 
 	"github.com/wso2/api-platform/gateway/gateway-runtime/policy-engine/internal/config"
 	"github.com/wso2/api-platform/gateway/gateway-runtime/policy-engine/internal/kernel"
@@ -332,3 +333,23 @@ func TestInitializeXDSClient_ValidConfig(t *testing.T) {
 	// Note: Not calling Stop/Wait due to potential issues with context in test environment
 	// The client will be cleaned up when the test exits
 }
+
+// The ext_proc server must be constructed with all three bounds set. The values
+// themselves are validated in internal/config; what this pins is that none of the three
+// options is dropped from the construction, which is how this server silently ran on
+// gRPC's defaults — a 4 MiB receive cap and unbounded streams — before.
+func TestExtProcServerOptions(t *testing.T) {
+	cfg := &config.Config{}
+	cfg.PolicyEngine.Server.MaxRecvMsgBytes = 11 << 20
+	cfg.PolicyEngine.Server.MaxSendMsgBytes = 11 << 20
+	cfg.PolicyEngine.Server.MaxConcurrentStreams = 4096
+
+	opts := extProcServerOptions(cfg)
+	assert.Len(t, opts, 3, "MaxRecvMsgSize, MaxSendMsgSize and MaxConcurrentStreams")
+
+	// And the real constructor accepts them, rather than this merely being a slice of
+	// the right length.
+	srv := grpc.NewServer(opts...)
+	require.NotNil(t, srv)
+	srv.Stop()
+}

gateway/gateway-runtime/policy-engine/internal/config/config.go

@@ -40,6 +40,26 @@ const (
 	// from a Content-Encoded body — the whole body when buffered, each chunk when
 	// streaming. Applied when max_decompressed_bytes is unset for a direction.
 	DefaultMaxDecompressedBytes int64 = 10 * 1024 * 1024 // 10 MiB
+
+	// ExtProcMessageOverheadBytes is the headroom an ext_proc message needs above the
+	// body it carries: request/response headers, Envoy attributes, dynamic metadata and
+	// protobuf framing all travel in the same message. The gRPC message limits are
+	// validated against the body ceilings plus this, because a limit sized to the body
+	// alone fails mid-request with ResourceExhausted on any request whose headers are
+	// large — a failure that looks like a gateway fault rather than a misconfiguration.
+	ExtProcMessageOverheadBytes int64 = 1 * 1024 * 1024 // 1 MiB
+
+	// DefaultMaxConcurrentStreams bounds in-flight ext_proc calls on the Envoy
+	// connection, one stream per request being processed. gRPC's own default is
+	// effectively unlimited, so an explicit value is what makes the stream budget a
+	// bounded resource rather than whatever the peer asks for.
+	//
+	// Deliberately generous. This is not a load-shedding control: Envoy does not
+	// degrade gracefully when it runs out of streams, it stalls, so a value below a
+	// pod's peak concurrent in-flight requests costs availability rather than
+	// protecting anything. Raise it if a single runtime instance legitimately carries
+	// more concurrency than this.
+	DefaultMaxConcurrentStreams uint32 = 10000
 )
 
 // defaultFileSourceAllowlist is the policy-engine's default set of directories that
@@ -239,6 +259,34 @@ type ServerConfig struct {
 
 	// ExtProcPort is the port for the ext_proc gRPC server (TCP mode only)
 	ExtProcPort int `koanf:"extproc_port"`
+
+	// MaxRecvMsgBytes and MaxSendMsgBytes bound one ext_proc message in each
+	// direction. Both must accommodate the larger of the two body decompression
+	// ceilings plus ExtProcMessageOverheadBytes, because both directions carry both
+	// kinds of body: the engine receives a request body and may return a mutated one,
+	// then receives a response body and may return a mutated one.
+	//
+	// They exist because gRPC's defaults are not this service's threat model — the
+	// receive default is 4 MiB regardless of how the body ceilings are configured, and
+	// the send default is unbounded.
+	MaxRecvMsgBytes int64 `koanf:"max_recv_msg_bytes"`
+	MaxSendMsgBytes int64 `koanf:"max_send_msg_bytes"`
+
+	// MaxConcurrentStreams bounds concurrent in-flight ext_proc calls. See
+	// DefaultMaxConcurrentStreams for why this is a generous bound rather than a
+	// load-shedding knob.
+	MaxConcurrentStreams uint32 `koanf:"max_concurrent_streams"`
+}
+
+// RequiredExtProcMessageBytes is the smallest message limit coherent with the
+// configured body ceilings. Both directions are sized off the larger ceiling, since
+// each carries request and response bodies alike.
+func (p PolicyEngine) RequiredExtProcMessageBytes() int64 {
+	ceiling := p.RequestBody.MaxDecompressedBytes
+	if p.ResponseBody.MaxDecompressedBytes > ceiling {
+		ceiling = p.ResponseBody.MaxDecompressedBytes
+	}
+	return ceiling + ExtProcMessageOverheadBytes
 }
 
 // PythonExecutorConfig holds configuration for the Python executor bridge.
@@ -511,8 +559,11 @@ func defaultConfig() *Config {
 	return &Config{
 		PolicyEngine: PolicyEngine{
 			Server: ServerConfig{
-				Mode:        "",
-				ExtProcPort: 9001,
+				Mode:                 "",
+				ExtProcPort:          9001,
+				MaxRecvMsgBytes:      DefaultMaxDecompressedBytes + ExtProcMessageOverheadBytes,
+				MaxSendMsgBytes:      DefaultMaxDecompressedBytes + ExtProcMessageOverheadBytes,
+				MaxConcurrentStreams: DefaultMaxConcurrentStreams,
 			},
 			Admin: AdminConfig{
 				Enabled:    true,
@@ -682,6 +733,54 @@ func (c *Config) Validate() error {
 		return fmt.Errorf("policy_engine.response_body.max_decompressed_bytes must be positive, got %d", c.PolicyEngine.ResponseBody.MaxDecompressedBytes)
 	}
 
+	// ext_proc gRPC message and stream limits.
+	//
+	// Unset means "derive from the body ceilings" rather than "reject", so a Config built
+	// in code (tests, embedders) stays usable and an operator who raises a body ceiling
+	// does not also have to restate the message limits. Load() starts from
+	// defaultConfig(), so a file-sourced config already carries values; this covers the
+	// rest. Same normalise-then-validate shape as the router's
+	// per_connection_buffer_limit_bytes.
+	required := c.PolicyEngine.RequiredExtProcMessageBytes()
+	if c.PolicyEngine.Server.MaxRecvMsgBytes == 0 {
+		c.PolicyEngine.Server.MaxRecvMsgBytes = required
+	}
+	if c.PolicyEngine.Server.MaxSendMsgBytes == 0 {
+		c.PolicyEngine.Server.MaxSendMsgBytes = required
+	}
+	if c.PolicyEngine.Server.MaxConcurrentStreams == 0 {
+		c.PolicyEngine.Server.MaxConcurrentStreams = DefaultMaxConcurrentStreams
+	}
+
+	// An *explicit* value below the ceiling is still refused: a message limit under the
+	// body a policy is allowed to buffer fails mid-request with ResourceExhausted, which
+	// surfaces as a gateway fault on live traffic instead of a startup error naming the
+	// two settings that disagree. Refusing to start is the cheaper failure.
+	if c.PolicyEngine.Server.MaxRecvMsgBytes < required {
+		return fmt.Errorf(
+			"policy_engine.server.max_recv_msg_bytes is %d, which is below the %d required by the configured "+
+				"body decompression ceilings plus %d of ext_proc message overhead",
+			c.PolicyEngine.Server.MaxRecvMsgBytes, required, ExtProcMessageOverheadBytes)
+	}
+	if c.PolicyEngine.Server.MaxSendMsgBytes < required {
+		return fmt.Errorf(
+			"policy_engine.server.max_send_msg_bytes is %d, which is below the %d required by the configured "+
+				"body decompression ceilings plus %d of ext_proc message overhead",
+			c.PolicyEngine.Server.MaxSendMsgBytes, required, ExtProcMessageOverheadBytes)
+	}
+	// grpc.MaxRecvMsgSize/MaxSendMsgSize take an int, so a value that does not survive
+	// the conversion would silently become a different limit than the one configured.
+	// int is 64-bit on every platform this ships on, making this unreachable there —
+	// which is the point of asserting it here rather than at the conversion.
+	for name, v := range map[string]int64{
+		"max_recv_msg_bytes": c.PolicyEngine.Server.MaxRecvMsgBytes,
+		"max_send_msg_bytes": c.PolicyEngine.Server.MaxSendMsgBytes,
+	} {
+		if int64(int(v)) != v {
+			return fmt.Errorf("policy_engine.server.%s is %d, which does not fit this platform's int", name, v)
+		}
+	}
+
 	// Validate config mode
 	if c.PolicyEngine.ConfigMode.Mode != "file" && c.PolicyEngine.ConfigMode.Mode != "xds" {
 		return fmt.Errorf("invalid config_mode.mode: %s (must be 'file' or 'xds')", c.PolicyEngine.ConfigMode.Mode)

gateway/gateway-runtime/policy-engine/internal/config/config_test.go

@@ -1654,3 +1654,73 @@ func TestDefaultConfig(t *testing.T) {
 	err := cfg.Validate()
 	assert.NoError(t, err)
 }
+
+// The ext_proc gRPC server carries every request and response body, so its message
+// limits have to be coherent with the body ceilings. Incoherent config must stop the
+// process at startup rather than fail mid-request with ResourceExhausted, which on live
+// traffic looks like a gateway fault instead of a setting that needs changing.
+func TestValidate_ExtProcMessageLimits(t *testing.T) {
+	baseline := func() *Config {
+		c := defaultConfig()
+		c.PolicyEngine.Server.Mode = "uds"
+		return c
+	}
+
+	t.Run("defaults are coherent with the default ceilings", func(t *testing.T) {
+		c := baseline()
+		require.NoError(t, c.Validate())
+		assert.Equal(t, DefaultMaxDecompressedBytes+ExtProcMessageOverheadBytes,
+			c.PolicyEngine.Server.MaxRecvMsgBytes)
+		assert.Equal(t, DefaultMaxConcurrentStreams, c.PolicyEngine.Server.MaxConcurrentStreams)
+	})
+
+	// Unset derives rather than rejects, so a Config built in code stays usable and an
+	// operator who raises a body ceiling need not restate the message limits.
+	t.Run("unset limits are derived from the ceilings", func(t *testing.T) {
+		c := baseline()
+		c.PolicyEngine.RequestBody.MaxDecompressedBytes = 50 * 1024 * 1024
+		c.PolicyEngine.Server.MaxRecvMsgBytes = 0
+		c.PolicyEngine.Server.MaxSendMsgBytes = 0
+		c.PolicyEngine.Server.MaxConcurrentStreams = 0
+
+		require.NoError(t, c.Validate())
+		want := int64(50*1024*1024) + ExtProcMessageOverheadBytes
+		assert.Equal(t, want, c.PolicyEngine.Server.MaxRecvMsgBytes)
+		assert.Equal(t, want, c.PolicyEngine.Server.MaxSendMsgBytes)
+		assert.Equal(t, DefaultMaxConcurrentStreams, c.PolicyEngine.Server.MaxConcurrentStreams)
+	})
+
+	// Both directions are sized off the *larger* ceiling, because each carries request
+	// and response bodies alike: the engine receives a request body and may return a
+	// mutated one, then receives a response body and may return a mutated one.
+	t.Run("the larger ceiling sets the requirement for both directions", func(t *testing.T) {
+		c := baseline()
+		c.PolicyEngine.RequestBody.MaxDecompressedBytes = 2 * 1024 * 1024
+		c.PolicyEngine.ResponseBody.MaxDecompressedBytes = 40 * 1024 * 1024
+
+		assert.Equal(t, int64(40*1024*1024)+ExtProcMessageOverheadBytes,
+			c.PolicyEngine.RequiredExtProcMessageBytes())
+	})
+
+	for name, mutate := range map[string]func(*Config){
+		"recv below the ceiling": func(c *Config) {
+			c.PolicyEngine.Server.MaxRecvMsgBytes = DefaultMaxDecompressedBytes
+		},
+		"send below the ceiling": func(c *Config) {
+			c.PolicyEngine.Server.MaxSendMsgBytes = DefaultMaxDecompressedBytes
+		},
+		"recv below a raised ceiling": func(c *Config) {
+			c.PolicyEngine.ResponseBody.MaxDecompressedBytes = 100 * 1024 * 1024
+		},
+	} {
+		t.Run("rejected: "+name, func(t *testing.T) {
+			c := baseline()
+			mutate(c)
+			err := c.Validate()
+			require.Error(t, err)
+			// The error has to name the two settings that disagree, or the operator is
+			// left guessing which of them to change.
+			assert.Regexp(t, `max_(recv|send)_msg_bytes is \d+, which is below the \d+ required`, err.Error())
+		})
+	}
+}

gateway/gateway-runtime/policy-engine/internal/kernel/execution_context.go

@@ -989,8 +989,9 @@ func (ec *PolicyExecutionContext) processResponseHeaders(
 	//   - The operation must have *declared* itself streaming. An Auto route is every kind
 	//     shipping today, and one with a buffered-only response policy meeting a chunked or
 	//     SSE upstream response has always simply buffered it. Turning that into a 500
-	//     would be a behavioural change for existing kinds, which §5.6 forbids outright —
-	//     the fail-closed behaviour is opt-in with the declaration, not a new global rule.
+	//     would be a behavioural change for existing kinds, which this must not introduce
+	//     — the fail-closed behaviour is opt-in with the declaration, not a new global
+	//     rule.
 	//   - responseNeedsStreaming looks at *this* response, not merely at the operation. A
 	//     streaming operation is entitled to answer with a 400 and a JSON error body, or a
 	//     bodyless 204; buffering those is correct, and the error the agent was trying to

gateway/gateway-runtime/policy-engine/internal/kernel/mapper.go

@@ -53,8 +53,10 @@ type RouteConfig struct {
 	// gateway hold: that is Envoy's listener-wide
 	// router.http_listener.per_connection_buffer_limit_bytes (1 MiB by default) and the
 	// ext_proc gRPC server's receive limit, neither of which is per-route. Lowering
-	// this value does not lower that. See §8 R3 in the design plan for the two
-	// mechanisms that would.
+	// this value does not lower that. Making it a real buffering bound needs either an
+	// Envoy-side per-route cap (the buffer filter's max_request_bytes, which returns 413
+	// before ext_proc collects the body) or streamed accumulation in the engine; neither
+	// is built.
 	MaxRequestBodyBytes int64
 }
 

gateway/gateway-runtime/policy-engine/internal/kernel/resolution.go

@@ -226,7 +226,7 @@ func (ec *PolicyExecutionContext) bindPendingChainAndProcess(
 	// collected the whole body and sent it here in one message by the time this runs.
 	// The memory an unauthenticated caller can pin is bounded by Envoy's listener-wide
 	// per_connection_buffer_limit_bytes and the ext_proc gRPC receive limit, neither of
-	// which is per-route. See RouteConfig.MaxRequestBodyBytes and §8 R3.
+	// which is per-route. See RouteConfig.MaxRequestBodyBytes.
 	if limit := pending.route.EffectiveMaxRequestBodyBytes(); int64(len(wire)) > limit {
 		return ec.denyResolution(ctx, &resolver.ResolutionError{
 			Kind: resolver.FailurePayloadTooLarge,