diff --git a/CHANGELOG.md b/CHANGELOG.md index 6b29f730e2..8c10fb26cf 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -37,6 +37,7 @@ Ref: https://keepachangelog.com/en/1.0.0/ * [#4009](https://github.com/sei-protocol/sei-chain/pull/4009) Bound `/store/*/subspace` ABCI queries with pair/byte caps, empty-prefix rejection, SS-path concurrency limits, and context-aware iteration to prevent memory-exhaustion DoS. * [#4032](https://github.com/sei-protocol/sei-chain/pull/4032) fix(config): the default `telemetry.prometheus-retention-time` drops from `7200` to `0`, so neither app.toml-generation pipeline (`seid init`, or the file a node writes for itself on any other subcommand) starts the Prometheus metrics sink unless an operator sets a positive retention. Freshly generated nodes keep the bounded in-memory telemetry sink used by SIGUSR1 dumps. Existing `app.toml` files are unchanged. * [#4021](https://github.com/sei-protocol/sei-chain/pull/4021) feat(grpc): per-IP rate-limit admission for the gRPC plane, off by default behind `[grpc] rate-limiting-enabled` (new `ip-rate-limit-rps` / `ip-rate-limit-burst` / `trusted-proxy-cidrs`, defaults 10 rps / 20 burst / trust no proxy). Native gRPC (:9090) is admitted by a tap handler and gRPC-Web (:9091) by HTTP middleware, both before the request is protobuf-decoded, so a throttled caller cannot spend the decoder; streams pay one token to establish and one per inbound message. Both planes draw from the same per-IP buckets. Over-budget callers get `ResourceExhausted` on :9090 and HTTP 429 on :9091, counted by `rpc_rate_limit_rejected_total{plane="grpc", method_namespace}`. +* [#4078](https://github.com/sei-protocol/sei-chain/pull/4078) feat(grpc): bound concurrent in-flight RPCs and open connections per IP on the gRPC query plane. New `[grpc] max-connections-per-ip` and `[grpc-web] max-connections-per-ip` (default 0, unlimited) optionally cap one address's share of the global connection budget on :9090 and :9091, regardless of `rate-limiting-enabled`. New `[grpc] max-in-flight-per-ip` (default 100) caps concurrent RPCs per address when `rate-limiting-enabled = true`: the slot is taken at the HTTP/2 HEADERS frame and returned when the RPC ends. Both planes draw from the same per-IP pool. Concurrency rejections return `ResourceExhausted` on :9090 and HTTP 429 on :9091, counted by `rpc_inflight_rejected_total{plane, method_namespace}`; refused connections are counted by `rpc_connection_rejected_total{plane}`. ### Upgrade guide * **IBC core removal.** Removes the retired IBC core source, protobufs, light clients, CLI, and simulation support. Retired IBC stores remain mounted but are omitted from `export-genesis`; preserve the state database or use v6.6 freeze nodes for historical IBC data. @@ -48,6 +49,7 @@ Ref: https://keepachangelog.com/en/1.0.0/ * [#4032](https://github.com/sei-protocol/sei-chain/pull/4032) **The Prometheus telemetry sink is off by default.** A node whose `app.toml` is generated by this release gets `prometheus-retention-time = 0`, which leaves the sink uncreated even though `telemetry.enabled` stays `true`. `GET /metrics?format=prometheus` on the app API server (:1317) then returns `prometheus metrics are not enabled`, and the `seid` process exports no application Prometheus series. **Operators who scrape application metrics should set a positive `[telemetry] prometheus-retention-time` in `app.toml` (the previous default was `7200`) before generating a new configuration file.** Nodes that already have `prometheus-retention-time` written in `app.toml` are unaffected. * [#3984](https://github.com/sei-protocol/sei-chain/pull/3984) **ABCI/gRPC pagination is now capped by default.** Untrusted callers requesting `limit` above 1000, `offset` above 10000, or a scan that exceeds 11000 total iterations now get `InvalidArgument` (over-cap) or a partial page with `next_key` (budget exhausted) instead of the previously unbounded scan. Clients that page with large limits/offsets, or trusted internal indexers, should either follow `next_key` for resumption or be added to the new `[query] trusted-cidrs` allowlist (or set `[query] disable-limits = true`) before upgrading. * [#4021](https://github.com/sei-protocol/sei-chain/pull/4021) **gRPC per-IP rate limiting defaults are deliberately conservative.** Admission stays off unless `[grpc] rate-limiting-enabled = true`, but the defaults it enables are 10 rps / 20 burst per IP. Streams pay one token to establish and one per inbound message, so at the default burst a client that sends more than ~20 messages in a burst is cut off mid-stream with `ResourceExhausted`. Operators enabling admission should size `ip-rate-limit-rps` / `ip-rate-limit-burst` against their heaviest streaming client, and set `trusted-proxy-cidrs` to their ingress CIDRs so callers are not all bucketed under the proxy's IP. +* [#4078](https://github.com/sei-protocol/sei-chain/pull/4078) **Optional per-IP connection and in-flight RPC caps on :9090 and :9091.** `max-connections-per-ip` defaults to 0 (unlimited) on both planes; set a positive value to cap one address's share of the global connection budget. The cap keys on the TCP peer that opened the socket and is unaffected by `trusted-proxy-cidrs`, which needs a request to read `X-Forwarded-For` from and so cannot apply at accept time: everything reaching the node over `127.0.0.1` shares one allowance, as does every client behind a proxy or load balancer, and IPv6 peers share per /64. **Operators exposing public gRPC query endpoints directly may want to set `[grpc] max-connections-per-ip` and `[grpc-web] max-connections-per-ip` (for example to 100, a tenth of the default global budget); on a node behind an ingress, size it against that ingress's total concurrent connections rather than one client's, or leave it at 0.** The companion `[grpc] max-in-flight-per-ip` (default 100 concurrent RPCs per address) only takes effect when `rate-limiting-enabled = true`; size it against your heaviest client's concurrency. * [#4009](https://github.com/sei-protocol/sei-chain/pull/4009) **`/store/*/subspace` scans are now capped.** Wide prefix scans that previously returned unbounded KV pairs now fail with `subspace result exceeds limit` once they would exceed the default caps of 1,000 pairs or 4 MiB of accumulated key+value bytes. Empty prefixes are rejected. Indexers and tooling that issue wide `/subspace` queries must narrow their prefixes, shard by sub-prefix, or raise `[state-commit] sc-subspace-max-pairs` and `sc-subspace-max-bytes` before upgrading. Values `<= 0` resolve to these defaults; there is no unlimited setting. * [#3927](https://github.com/sei-protocol/sei-chain/pull/3927) **Legacy Sei JSON-RPC and CLI removal.** Removes `sei_associate`, `sei_getBlockByHash`, `sei_getBlockByHashExcludeTraceFail`, `sei_getBlockTransactionCountByHash`, `sei_getBlockTransactionCountByNumber`, `sei_getEvmTx`, `sei_getFilterChanges`, `sei_getFilterLogs`, `sei_getLogs`, `sei_getTransactionByBlockHashAndIndex`, `sei_getTransactionByBlockNumberAndIndex`, `sei_getTransactionByHash`, `sei_getTransactionCount`, `sei_getTransactionErrorByHash`, `sei_getTransactionReceiptExcludeTraceFail`, `sei_getVMError`, `sei_newBlockFilter`, `sei_newFilter`, `sei_sign`, and `sei_uninstallFilter`. Use standard `eth_*` methods for EVM-originated data and `seid tx evm native-associate -y` for address association. There is no block- or filter-level replacement for discovering Cosmos-originated synthetic logs; clients that know the synthetic transaction hash can enable `sei_getTransactionReceipt`. diff --git a/config/cosmosbase/agreement_test.go b/config/cosmosbase/agreement_test.go index 0fd97eb226..effffb0074 100644 --- a/config/cosmosbase/agreement_test.go +++ b/config/cosmosbase/agreement_test.go @@ -102,6 +102,7 @@ func readerValues(t *testing.T) map[string]string { "grpc.address": fmt.Sprint(cfg.GRPC.Address), "grpc.max-recv-msg-size": fmt.Sprint(cfg.GRPC.MaxRecvMsgSize), "grpc.max-open-connections": fmt.Sprint(cfg.GRPC.MaxOpenConnections), + "grpc.max-connections-per-ip": fmt.Sprint(cfg.GRPC.MaxConnectionsPerIP), "grpc.max-connection-idle": fmt.Sprint(cfg.GRPC.MaxConnectionIdle), "grpc.max-connection-age": fmt.Sprint(cfg.GRPC.MaxConnectionAge), "grpc.max-connection-age-grace": fmt.Sprint(cfg.GRPC.MaxConnectionAgeGrace), @@ -111,6 +112,7 @@ func readerValues(t *testing.T) map[string]string { "grpc.keepalive-permit-without-stream": fmt.Sprint(cfg.GRPC.KeepalivePermitWithoutStream), "grpc.ip-rate-limit-rps": fmt.Sprint(cfg.GRPC.IPRateLimitRPS), "grpc.ip-rate-limit-burst": fmt.Sprint(cfg.GRPC.IPRateLimitBurst), + "grpc.max-in-flight-per-ip": fmt.Sprint(cfg.GRPC.MaxInFlightPerIP), "grpc.rate-limiting-enabled": fmt.Sprint(cfg.GRPC.RateLimitingEnabled), "grpc.trusted-proxy-cidrs": fmt.Sprint(cfg.GRPC.TrustedProxyCIDRs), "telemetry.service-name": fmt.Sprint(cfg.Telemetry.ServiceName), diff --git a/config/cosmosbase/cosmosbase.go b/config/cosmosbase/cosmosbase.go index ba37836a12..b7d7a2e90d 100644 --- a/config/cosmosbase/cosmosbase.go +++ b/config/cosmosbase/cosmosbase.go @@ -96,10 +96,11 @@ func apiDefaults(mode registry.Mode) any { return forMode(mode).API } // interface follows and for the same reason. The upstream default is on for every kind, so declaring that // would state an open interface on the nodes meant to expose the least. // -// Nine of these fifteen keys are read only when the key is present. Two more are durations read through a -// clamp that rescues a negative value and does nothing for an absent one, so those two are unguarded and -// their clobber leaves no trace. The durations are declared as durations and written into a file as text, -// which is the shape the reader parses back. +// Most gRPC settings preserve their in-code defaults when the key is absent because the reader only reads +// them after an IsSet check. The connection-age durations are the exception: they are read unconditionally +// and clamped only when negative. Their default is zero, and Viper also returns zero for an absent duration, +// so an omitted key resolves to the same value as the declared default. Duration defaults are rendered as +// text in the config file, matching the shape the reader parses back. func grpcDefaults(mode registry.Mode) any { return forMode(mode).GRPC } // stateSyncDefaults is what the snapshot settings resolve to for a node of this kind. diff --git a/config/cosmosbase/cosmosbase_test.go b/config/cosmosbase/cosmosbase_test.go index 29ff424f3b..56fcf5725a 100644 --- a/config/cosmosbase/cosmosbase_test.go +++ b/config/cosmosbase/cosmosbase_test.go @@ -74,11 +74,12 @@ func TestTheRESTKeysAreTheOnesItsReaderResolves(t *testing.T) { func TestTheGRPCKeysAreTheOnesItsReaderResolves(t *testing.T) { requireDeclares(t, GRPCSectionName, []string{ "grpc.enable", "grpc.address", "grpc.max-recv-msg-size", "grpc.max-open-connections", + "grpc.max-connections-per-ip", "grpc.max-connection-idle", "grpc.max-connection-age", "grpc.max-connection-age-grace", "grpc.keepalive-time", "grpc.keepalive-timeout", "grpc.keepalive-min-time", "grpc.keepalive-permit-without-stream", - "grpc.ip-rate-limit-rps", "grpc.ip-rate-limit-burst", "grpc.rate-limiting-enabled", - "grpc.trusted-proxy-cidrs", + "grpc.ip-rate-limit-rps", "grpc.ip-rate-limit-burst", "grpc.max-in-flight-per-ip", + "grpc.rate-limiting-enabled", "grpc.trusted-proxy-cidrs", }) } diff --git a/ratelimiter/conn_limit.go b/ratelimiter/conn_limit.go new file mode 100644 index 0000000000..f9fd7619bd --- /dev/null +++ b/ratelimiter/conn_limit.go @@ -0,0 +1,88 @@ +package ratelimiter + +import ( + "context" + "net" + "sync" + + "go.opentelemetry.io/otel/attribute" + "go.opentelemetry.io/otel/metric" +) + +// ConnLimitListener returns a listener that bounds the number of simultaneously +// open connections from any one client address to maxPerIP. A non-positive +// maxPerIP returns inner unchanged. +// +// It bounds the layer below the RPC — accepted sockets, TLS handshakes, HTTP/2 +// frame state — which a per-RPC counter cannot see, and it caps the per-address +// share of the global connection budget. Wrap the raw listener with this before +// the global cap, so a connection this rejects never spends a global slot. +// +// Addresses are keyed on the TCP peer, normalized the way rate-limit buckets +// are: a client rotating within an IPv6 /64 does not get a fresh allowance per +// address, and every client sharing an address shares one allowance. +// trusted-proxy-cidrs does not apply here, so on a node behind a proxy or load +// balancer maxPerIP bounds that proxy as a whole rather than its clients. +func ConnLimitListener(inner net.Listener, plane string, maxPerIP int) net.Listener { + if maxPerIP <= 0 { + return inner + } + return &connLimitListener{ + Listener: inner, + plane: plane, + counter: newInflightCounter(maxPerIP), + } +} + +// connLimitListener is the listener ConnLimitListener returns. +type connLimitListener struct { + net.Listener + + plane string + counter *inflightCounter +} + +// Accept returns the next connection whose client address is under the cap, +// closing and counting the ones that are not. +// +// Over-cap connections are dropped here rather than surfaced as an Accept error, +// because an Accept error stops the serving loop and would turn one client's +// excess into an outage for every other client. +func (l *connLimitListener) Accept() (net.Conn, error) { + for { + conn, err := l.Listener.Accept() + if err != nil { + return nil, err + } + // The peer address is all a listener has: nothing has been read off the + // socket yet, so there is no X-Forwarded-For to resolve a client behind + // a trusted proxy to. + key := bucketKey(stripPort(conn.RemoteAddr().String())) + if l.counter.acquire(key) { + return &limitedConn{Conn: conn, release: func() { l.counter.release(key) }}, nil + } + registryMetrics.connRejectedCounter.Add( + context.Background(), + 1, + metric.WithAttributes(attribute.String("plane", l.plane)), + ) + _ = conn.Close() + } +} + +// limitedConn returns its address's connection slot when it is closed. +type limitedConn struct { + net.Conn + + once sync.Once + release func() +} + +// Close closes the underlying connection and returns its slot. It is safe to +// call more than once: net/http and grpc-go both close a connection from more +// than one path, and a double release would hand the address a free slot. +func (c *limitedConn) Close() error { + err := c.Conn.Close() + c.once.Do(c.release) + return err +} diff --git a/ratelimiter/conn_limit_test.go b/ratelimiter/conn_limit_test.go new file mode 100644 index 0000000000..361d0e0642 --- /dev/null +++ b/ratelimiter/conn_limit_test.go @@ -0,0 +1,118 @@ +package ratelimiter + +import ( + "io" + "net" + "testing" + "time" + + "github.com/stretchr/testify/require" +) + +// acceptedConns runs a listener's Accept loop into a channel so a test can see +// which connections the per-IP cap let through. +func acceptedConns(t *testing.T, ln net.Listener) <-chan net.Conn { + t.Helper() + out := make(chan net.Conn, 16) + go func() { + defer close(out) + for { + conn, err := ln.Accept() + if err != nil { + return + } + out <- conn + } + }() + return out +} + +func dialTo(t *testing.T, ln net.Listener) net.Conn { + t.Helper() + conn, err := net.Dial("tcp", ln.Addr().String()) + require.NoError(t, err) + t.Cleanup(func() { _ = conn.Close() }) + return conn +} + +// requireClosedByServer pins that the server hung up on conn rather than serving +// it, which is how an over-cap connection is refused. +func requireClosedByServer(t *testing.T, conn net.Conn) { + t.Helper() + require.NoError(t, conn.SetReadDeadline(time.Now().Add(5*time.Second))) + _, err := conn.Read(make([]byte, 1)) + require.ErrorIs(t, err, io.EOF) +} + +func TestConnLimitListenerCapsConnectionsPerIP(t *testing.T) { + raw, err := net.Listen("tcp", "127.0.0.1:0") + require.NoError(t, err) + ln := ConnLimitListener(raw, PlaneGRPC, 2) + t.Cleanup(func() { _ = ln.Close() }) + accepted := acceptedConns(t, ln) + + first := dialTo(t, ln) + second := dialTo(t, ln) + served := []net.Conn{<-accepted, <-accepted} + + // The third is dropped, and the loop keeps serving rather than failing. + requireClosedByServer(t, dialTo(t, ln)) + require.Empty(t, accepted) + + // Closing a served connection returns its slot. + require.NoError(t, served[0].Close()) + require.NoError(t, first.Close()) + dialTo(t, ln) + require.NotNil(t, <-accepted) + + _ = second.Close() + _ = served[1].Close() +} + +// TestConnLimitListenerDoubleCloseReturnsOneSlot pins the idempotent release: +// net/http and grpc-go both close a connection from more than one path, and a +// second release would hand the address a slot it is not holding. +func TestConnLimitListenerDoubleCloseReturnsOneSlot(t *testing.T) { + raw, err := net.Listen("tcp", "127.0.0.1:0") + require.NoError(t, err) + ln := ConnLimitListener(raw, PlaneGRPC, 1) + t.Cleanup(func() { _ = ln.Close() }) + accepted := acceptedConns(t, ln) + + dialTo(t, ln) + served := <-accepted + require.NoError(t, served.Close()) + _ = served.Close() + + limiter, ok := ln.(*connLimitListener) + require.True(t, ok) + require.Equal(t, 0, limiter.counter.heldFor("127.0.0.1")) + + // One slot came back, not two. + dialTo(t, ln) + require.NotNil(t, <-accepted) + requireClosedByServer(t, dialTo(t, ln)) +} + +// TestConnLimitListenerDisabledReturnsInner pins that a non-positive cap adds no +// wrapper at all, so the listener behaves exactly as before. +func TestConnLimitListenerDisabledReturnsInner(t *testing.T) { + raw, err := net.Listen("tcp", "127.0.0.1:0") + require.NoError(t, err) + t.Cleanup(func() { _ = raw.Close() }) + + require.Same(t, raw, ConnLimitListener(raw, PlaneGRPC, 0)) + require.Same(t, raw, ConnLimitListener(raw, PlaneGRPC, -1)) +} + +// TestConnLimitListenerAcceptStopsOnListenerError pins that a closed listener +// still ends the Accept loop rather than spinning inside it. +func TestConnLimitListenerAcceptStopsOnListenerError(t *testing.T) { + raw, err := net.Listen("tcp", "127.0.0.1:0") + require.NoError(t, err) + ln := ConnLimitListener(raw, PlaneGRPC, 1) + require.NoError(t, ln.Close()) + + _, err = ln.Accept() + require.Error(t, err) +} diff --git a/ratelimiter/inflight.go b/ratelimiter/inflight.go new file mode 100644 index 0000000000..a826a667dd --- /dev/null +++ b/ratelimiter/inflight.go @@ -0,0 +1,110 @@ +package ratelimiter + +import ( + "context" + "sync" + + "go.opentelemetry.io/otel/attribute" + "go.opentelemetry.io/otel/metric" +) + +// inflightCounter bounds the number of slots held concurrently per key. +// +// A key is dropped once its count returns to zero, so the map is bounded by the +// number of slots actually held rather than by the number of addresses seen. +type inflightCounter struct { + max int + + mu sync.Mutex + held map[string]int +} + +func newInflightCounter(max int) *inflightCounter { + return &inflightCounter{max: max, held: make(map[string]int)} +} + +// acquire takes one slot for key and reports whether it was available. +func (c *inflightCounter) acquire(key string) bool { + c.mu.Lock() + defer c.mu.Unlock() + if c.held[key] >= c.max { + return false + } + c.held[key]++ + return true +} + +// release returns one slot for key. Releasing a key holding none is a no-op. +func (c *inflightCounter) release(key string) { + c.mu.Lock() + defer c.mu.Unlock() + n, ok := c.held[key] + if !ok { + return + } + if n <= 1 { + delete(c.held, key) + return + } + c.held[key] = n - 1 +} + +// heldFor returns the number of slots key currently holds. +func (c *inflightCounter) heldFor(key string) int { + c.mu.Lock() + defer c.mu.Unlock() + return c.held[key] +} + +// AcquireInFlight takes one concurrency slot for ip and reports whether one was +// available. It returns true when the Registry has no in-flight limit, so a +// caller pairs it with ReleaseInFlight unconditionally. +// +// Rejections increment rpc_inflight_rejected_total{plane, method_namespace}, +// the concurrency sibling of rpc_rate_limit_rejected_total. +func (r *Registry) AcquireInFlight(ctx context.Context, ip, plane, method string) bool { + if r.inflight == nil { + return true + } + if r.inflight.acquire(bucketKey(ip)) { + return true + } + registryMetrics.inflightRejectedCounter.Add( + ctx, + 1, + metric.WithAttributes( + attribute.String("plane", plane), + attribute.String("method_namespace", bucketRPCMethod(plane, method, r.knownGRPCMethods())), + ), + ) + return false +} + +// ReleaseInFlight returns the concurrency slot AcquireInFlight took for ip. +func (r *Registry) ReleaseInFlight(ip string) { + if r.inflight == nil { + return + } + r.inflight.release(bucketKey(ip)) +} + +// InFlightHeld returns the number of concurrency slots ip currently holds, and 0 +// when the Registry has no in-flight limit. +func (r *Registry) InFlightHeld(ip string) int { + if r.inflight == nil { + return 0 + } + return r.inflight.heldFor(bucketKey(ip)) +} + +// IsKnownGRPCMethod reports whether fullMethod is one of the "service/Method" +// names given to SetKnownGRPCMethods, with or without a leading slash. It +// reports false when SetKnownGRPCMethods has not been called. +func (r *Registry) IsKnownGRPCMethod(fullMethod string) bool { + known := r.knownGRPCMethods() + if known == nil { + return false + } + _, ok := known[trimLeadingSlash(fullMethod)] + return ok +} diff --git a/ratelimiter/inflight_test.go b/ratelimiter/inflight_test.go new file mode 100644 index 0000000000..6d88dbf30c --- /dev/null +++ b/ratelimiter/inflight_test.go @@ -0,0 +1,118 @@ +package ratelimiter + +import ( + "context" + "sync" + "testing" + + "github.com/stretchr/testify/require" +) + +func inFlightRegistry(t *testing.T, maxInFlight int) *Registry { + t.Helper() + r, err := New(Config{RPS: DefaultRPS, Burst: DefaultBurst, MaxInFlightPerIP: maxInFlight}) + require.NoError(t, err) + return r +} + +func TestAcquireInFlightCapsConcurrencyPerIP(t *testing.T) { + r := inFlightRegistry(t, 2) + ctx := context.Background() + + require.True(t, r.AcquireInFlight(ctx, "1.2.3.4", PlaneGRPC, "svc/M")) + require.True(t, r.AcquireInFlight(ctx, "1.2.3.4", PlaneGRPC, "svc/M")) + require.False(t, r.AcquireInFlight(ctx, "1.2.3.4", PlaneGRPC, "svc/M")) + + // The cap is per address, so a second IP has its own allowance. + require.True(t, r.AcquireInFlight(ctx, "5.6.7.8", PlaneGRPC, "svc/M")) + + r.ReleaseInFlight("1.2.3.4") + require.True(t, r.AcquireInFlight(ctx, "1.2.3.4", PlaneGRPC, "svc/M")) +} + +// TestReleaseInFlightDropsTheKey pins that the counter cannot grow with the +// number of addresses seen: an address holding nothing occupies no entry. +func TestReleaseInFlightDropsTheKey(t *testing.T) { + r := inFlightRegistry(t, 2) + ctx := context.Background() + + require.True(t, r.AcquireInFlight(ctx, "1.2.3.4", PlaneGRPC, "svc/M")) + require.True(t, r.AcquireInFlight(ctx, "1.2.3.4", PlaneGRPC, "svc/M")) + r.ReleaseInFlight("1.2.3.4") + require.Equal(t, 1, r.InFlightHeld("1.2.3.4")) + r.ReleaseInFlight("1.2.3.4") + + require.Equal(t, 0, r.InFlightHeld("1.2.3.4")) + require.Empty(t, r.inflight.held) +} + +// TestReleaseInFlightUnheldIsANoOp pins that an unmatched release cannot mint a +// slot, which would let an address exceed the cap. +func TestReleaseInFlightUnheldIsANoOp(t *testing.T) { + r := inFlightRegistry(t, 1) + ctx := context.Background() + + r.ReleaseInFlight("1.2.3.4") + r.ReleaseInFlight("1.2.3.4") + + require.True(t, r.AcquireInFlight(ctx, "1.2.3.4", PlaneGRPC, "svc/M")) + require.False(t, r.AcquireInFlight(ctx, "1.2.3.4", PlaneGRPC, "svc/M")) +} + +// TestAcquireInFlightSharesAnIPv6Prefix holds the concurrency cap to the same +// keying as the buckets: rotating within a /64 does not buy a fresh allowance. +func TestAcquireInFlightSharesAnIPv6Prefix(t *testing.T) { + r := inFlightRegistry(t, 1) + ctx := context.Background() + + require.True(t, r.AcquireInFlight(ctx, "2001:db8::1", PlaneGRPC, "svc/M")) + require.False(t, r.AcquireInFlight(ctx, "2001:db8::2", PlaneGRPC, "svc/M")) + require.True(t, r.AcquireInFlight(ctx, "2001:db9::1", PlaneGRPC, "svc/M")) +} + +// TestAcquireInFlightDisabledAdmitsEverything pins that a zero limit leaves the +// token bucket as the only admission control rather than blocking every RPC. +func TestAcquireInFlightDisabledAdmitsEverything(t *testing.T) { + r := inFlightRegistry(t, 0) + ctx := context.Background() + + for range 100 { + require.True(t, r.AcquireInFlight(ctx, "1.2.3.4", PlaneGRPC, "svc/M")) + } + require.Equal(t, 0, r.InFlightHeld("1.2.3.4")) + r.ReleaseInFlight("1.2.3.4") +} + +func TestAcquireInFlightIsConcurrencySafe(t *testing.T) { + const max = 8 + r := inFlightRegistry(t, max) + ctx := context.Background() + + var wg sync.WaitGroup + for range 64 { + wg.Add(1) + go func() { + defer wg.Done() + if r.AcquireInFlight(ctx, "1.2.3.4", PlaneGRPC, "svc/M") { + r.ReleaseInFlight("1.2.3.4") + } + }() + } + wg.Wait() + + require.Equal(t, 0, r.InFlightHeld("1.2.3.4")) +} + +func TestIsKnownGRPCMethod(t *testing.T) { + r := inFlightRegistry(t, 1) + + // Nothing is known until the server has declared what it serves. + require.False(t, r.IsKnownGRPCMethod("/testdata.Query/Echo")) + + r.SetKnownGRPCMethods([]string{"testdata.Query/Echo"}) + require.True(t, r.IsKnownGRPCMethod("/testdata.Query/Echo")) + require.True(t, r.IsKnownGRPCMethod("testdata.Query/Echo")) + require.False(t, r.IsKnownGRPCMethod("/testdata.Query/Nope")) + require.False(t, r.IsKnownGRPCMethod("/nope")) + require.False(t, r.IsKnownGRPCMethod("")) +} diff --git a/ratelimiter/method_bucket.go b/ratelimiter/method_bucket.go index a164f6454f..41fa1f27fb 100644 --- a/ratelimiter/method_bucket.go +++ b/ratelimiter/method_bucket.go @@ -7,6 +7,11 @@ const ( PlaneCometBFT = "cometbft" // PlaneGRPC is the rate-limit plane label for native gRPC (:9090). PlaneGRPC = "grpc" + // PlaneGRPCWeb is the plane label for gRPC-Web (:9091) connection + // rejections. The two planes share PlaneGRPC everywhere they share a + // per-IP pool; the connection cap is configured and enforced per listener, + // so its metric names the listener that refused. + PlaneGRPCWeb = "grpc-web" // rpcMethodBucketOther is the fallback label for unrecognized methods. rpcMethodBucketOther = "other" diff --git a/ratelimiter/metrics.go b/ratelimiter/metrics.go index 6f2371afc6..82647bcb4d 100644 --- a/ratelimiter/metrics.go +++ b/ratelimiter/metrics.go @@ -9,13 +9,25 @@ var ( registryMeter = otel.Meter("ratelimiter") registryMetrics = struct { - rejectedCounter metric.Int64Counter + rejectedCounter metric.Int64Counter + inflightRejectedCounter metric.Int64Counter + connRejectedCounter metric.Int64Counter }{ rejectedCounter: must(registryMeter.Int64Counter( "rpc_rate_limit_rejected_total", metric.WithDescription("Total RPC requests rejected by the per-IP rate limiter"), metric.WithUnit("{request}"), )), + inflightRejectedCounter: must(registryMeter.Int64Counter( + "rpc_inflight_rejected_total", + metric.WithDescription("Total RPC requests rejected by the per-IP concurrency limit"), + metric.WithUnit("{request}"), + )), + connRejectedCounter: must(registryMeter.Int64Counter( + "rpc_connection_rejected_total", + metric.WithDescription("Total connections rejected by the per-IP connection limit"), + metric.WithUnit("{connection}"), + )), } ) diff --git a/ratelimiter/registry.go b/ratelimiter/registry.go index dc2c2c9c2c..537e102c2e 100644 --- a/ratelimiter/registry.go +++ b/ratelimiter/registry.go @@ -56,6 +56,10 @@ type Config struct { // TrustedProxyCIDRs lists CIDRs whose X-Forwarded-For headers are trusted. // Empty means trust no proxy; use RemoteAddr / peer address directly. TrustedProxyCIDRs []string + // MaxInFlightPerIP is the number of RPCs one IP may have in flight at once. + // Zero disables the concurrency limit (AcquireInFlight always returns true), + // leaving the token bucket as the only admission control. + MaxInFlightPerIP int } // DefaultConfig uses no trusted proxies. If your node is behind a reverse proxy or @@ -74,6 +78,7 @@ type Registry struct { lru *expirable.LRU[string, *rate.Limiter] mu sync.Mutex grpcMethods atomic.Pointer[map[string]struct{}] + inflight *inflightCounter } // SetKnownGRPCMethods bounds the method label recorded for PlaneGRPC rejections @@ -82,7 +87,7 @@ type Registry struct { func (r *Registry) SetKnownGRPCMethods(methods []string) { known := make(map[string]struct{}, len(methods)) for _, m := range methods { - known[strings.TrimPrefix(m, "/")] = struct{}{} + known[trimLeadingSlash(m)] = struct{}{} } r.grpcMethods.Store(&known) } @@ -100,10 +105,15 @@ func New(cfg Config) (*Registry, error) { if err != nil { return nil, err } + var inflight *inflightCounter + if cfg.MaxInFlightPerIP > 0 { + inflight = newInflightCounter(cfg.MaxInFlightPerIP) + } return &Registry{ cfg: cfg, trustedProxies: proxies, lru: expirable.NewLRU[string, *rate.Limiter](lruSize, nil, lruTTL), + inflight: inflight, }, nil } @@ -257,6 +267,11 @@ func parseCIDRs(cidrs []string) ([]*net.IPNet, error) { return out, nil } +// trimLeadingSlash returns the gRPC "service/Method" name of a full method name. +func trimLeadingSlash(fullMethod string) string { + return strings.TrimPrefix(fullMethod, "/") +} + func stripPort(addr string) string { host, _, err := net.SplitHostPort(addr) if err != nil { diff --git a/sei-cosmos/server/config/config.go b/sei-cosmos/server/config/config.go index 852704be74..70db7578e3 100644 --- a/sei-cosmos/server/config/config.go +++ b/sei-cosmos/server/config/config.go @@ -32,10 +32,20 @@ const ( // simultaneous open connections for the gRPC-web server. DefaultGRPCWebMaxOpenConnections = 1000 + // DefaultGRPCWebMaxConnectionsPerIP defines the default maximum number of + // simultaneous open connections one client address may hold on the gRPC-web + // server. 0 means unlimited. + DefaultGRPCWebMaxConnectionsPerIP = 0 + // DefaultGRPCMaxOpenConnections defines the default maximum number of // simultaneous open connections for the gRPC server. 0 means unlimited. DefaultGRPCMaxOpenConnections = 1000 + // DefaultGRPCMaxConnectionsPerIP defines the default maximum number of + // simultaneous open connections one client address may hold on the gRPC + // server. 0 means unlimited. + DefaultGRPCMaxConnectionsPerIP = 0 + // DefaultGRPCMaxRecvMsgSize defines the default maximum message size in bytes // that the gRPC server can receive (4 MB), mirroring gRPC's own default. DefaultGRPCMaxRecvMsgSize = 4 * 1024 * 1024 @@ -83,6 +93,17 @@ const ( // the gRPC plane. DefaultGRPCIPRateLimitBurst = 20 + // DefaultGRPCMaxInFlightPerIP is the default number of RPCs one client + // address may have in flight at once on the gRPC plane. + // + // Five times the burst, so a client spending its whole bucket at once is not + // throttled by this instead. The bound it buys is on simultaneous decodes: + // with the default 4 MB message ceiling, one address can hold at most + // 100 x 4 MB of request buffers, where before it was capped only by the + // per-connection stream limit multiplied by the global connection budget. + // 0 means unlimited. + DefaultGRPCMaxInFlightPerIP = 100 + // DefaultOccEanbled defines whether to use OCC for tx processing DefaultOccEnabled = true ) @@ -240,6 +261,12 @@ type GRPCConfig struct { // connections. 0 means unlimited. MaxOpenConnections uint `mapstructure:"max-open-connections"` + // MaxConnectionsPerIP defines the maximum number of simultaneous open + // connections one client address may hold. It bounds the accepted sockets and + // HTTP/2 frame state below the RPC layer, which max-in-flight-per-ip cannot + // see. 0 means unlimited. + MaxConnectionsPerIP uint `mapstructure:"max-connections-per-ip"` + // MaxConnectionIdle is the duration after which an idle connection is closed. // 0 means infinity. MaxConnectionIdle time.Duration `mapstructure:"max-connection-idle"` @@ -278,6 +305,13 @@ type GRPCConfig struct { // admission when rate-limiting-enabled is true. IPRateLimitBurst int `mapstructure:"ip-rate-limit-burst"` + // MaxInFlightPerIP is the number of RPCs one client address may have in + // flight at once. An RPC is in flight from the moment its headers arrive + // until it ends, so this bounds concurrency where the token bucket bounds + // only arrival rate. Zero disables the limit. It takes effect only when + // rate-limiting-enabled is true. + MaxInFlightPerIP int `mapstructure:"max-in-flight-per-ip"` + // RateLimitingEnabled is the master switch for gRPC rate-limit admission. It // governs gRPC-Web (:9091) as well as native gRPC (:9090), and both planes // draw from the same per-IP buckets. @@ -295,6 +329,7 @@ func (c GRPCConfig) RateLimiterConfig() ratelimiter.Config { RPS: c.IPRateLimitRPS, Burst: c.IPRateLimitBurst, TrustedProxyCIDRs: c.TrustedProxyCIDRs, + MaxInFlightPerIP: c.MaxInFlightPerIP, } } @@ -311,6 +346,10 @@ type GRPCWebConfig struct { // MaxOpenConnections defines the maximum number of simultaneous open connections. 0 means unlimited. MaxOpenConnections uint `mapstructure:"max-open-connections"` + + // MaxConnectionsPerIP defines the maximum number of simultaneous open + // connections one client address may hold. 0 means unlimited. + MaxConnectionsPerIP uint `mapstructure:"max-connections-per-ip"` } // StateSyncConfig defines the state sync snapshot configuration. @@ -417,6 +456,7 @@ func DefaultConfig() *Config { Address: DefaultGRPCAddress, MaxRecvMsgSize: DefaultGRPCMaxRecvMsgSize, MaxOpenConnections: DefaultGRPCMaxOpenConnections, + MaxConnectionsPerIP: DefaultGRPCMaxConnectionsPerIP, MaxConnectionIdle: DefaultGRPCMaxConnectionIdle, MaxConnectionAge: DefaultGRPCMaxConnectionAge, MaxConnectionAgeGrace: DefaultGRPCMaxConnectionAgeGrace, @@ -426,6 +466,7 @@ func DefaultConfig() *Config { KeepalivePermitWithoutStream: DefaultGRPCKeepalivePermitWithoutStream, IPRateLimitRPS: DefaultGRPCIPRateLimitRPS, IPRateLimitBurst: DefaultGRPCIPRateLimitBurst, + MaxInFlightPerIP: DefaultGRPCMaxInFlightPerIP, RateLimitingEnabled: false, TrustedProxyCIDRs: nil, }, @@ -438,9 +479,10 @@ func DefaultConfig() *Config { Offline: false, }, GRPCWeb: GRPCWebConfig{ - Enable: true, - Address: DefaultGRPCWebAddress, - MaxOpenConnections: DefaultGRPCWebMaxOpenConnections, + Enable: true, + Address: DefaultGRPCWebAddress, + MaxOpenConnections: DefaultGRPCWebMaxOpenConnections, + MaxConnectionsPerIP: DefaultGRPCWebMaxConnectionsPerIP, }, Query: DefaultQueryConfig(), StateSync: StateSyncConfig{ @@ -581,10 +623,14 @@ func GetConfig(v *viper.Viper) (Config, error) { if v.IsSet("grpc-web.max-open-connections") { grpcWebMaxOpenConnections = v.GetUint("grpc-web.max-open-connections") } + grpcWebMaxConnectionsPerIP := uint(DefaultGRPCWebMaxConnectionsPerIP) + if v.IsSet("grpc-web.max-connections-per-ip") { + grpcWebMaxConnectionsPerIP = v.GetUint("grpc-web.max-connections-per-ip") + } // Apply in-code defaults when keys are absent so that nodes upgrading with an - // older app.toml (which lacks these keys) remain bounded rather than running - // with unlimited connections / message sizes. + // older app.toml (which lacks these keys) pick up the declared defaults rather + // than running with zero values where a limit is intended. grpcMaxRecvMsgSize := DefaultGRPCMaxRecvMsgSize if v.IsSet("grpc.max-recv-msg-size") { grpcMaxRecvMsgSize = v.GetInt("grpc.max-recv-msg-size") @@ -593,6 +639,10 @@ func GetConfig(v *viper.Viper) (Config, error) { if v.IsSet("grpc.max-open-connections") { grpcMaxOpenConnections = v.GetUint("grpc.max-open-connections") } + grpcMaxConnectionsPerIP := uint(DefaultGRPCMaxConnectionsPerIP) + if v.IsSet("grpc.max-connections-per-ip") { + grpcMaxConnectionsPerIP = v.GetUint("grpc.max-connections-per-ip") + } // Clamp negative durations back to their in-code defaults. A negative // keepalive/connection-age value is a misconfiguration that gRPC would // otherwise accept verbatim, so fall back to the safe default instead. @@ -625,6 +675,10 @@ func GetConfig(v *viper.Viper) (Config, error) { if v.IsSet("grpc.ip-rate-limit-burst") { grpcIPRateLimitBurst = v.GetInt("grpc.ip-rate-limit-burst") } + grpcMaxInFlightPerIP := DefaultGRPCMaxInFlightPerIP + if v.IsSet("grpc.max-in-flight-per-ip") { + grpcMaxInFlightPerIP = v.GetInt("grpc.max-in-flight-per-ip") + } grpcTrustedProxyCIDRs := []string(nil) if v.IsSet("grpc.trusted-proxy-cidrs") { grpcTrustedProxyCIDRs = v.GetStringSlice("grpc.trusted-proxy-cidrs") @@ -678,6 +732,7 @@ func GetConfig(v *viper.Viper) (Config, error) { Address: v.GetString("grpc.address"), MaxRecvMsgSize: grpcMaxRecvMsgSize, MaxOpenConnections: grpcMaxOpenConnections, + MaxConnectionsPerIP: grpcMaxConnectionsPerIP, MaxConnectionIdle: grpcMaxConnectionIdle, MaxConnectionAge: grpcMaxConnectionAge, MaxConnectionAgeGrace: grpcMaxConnectionAgeGrace, @@ -687,14 +742,16 @@ func GetConfig(v *viper.Viper) (Config, error) { KeepalivePermitWithoutStream: v.GetBool("grpc.keepalive-permit-without-stream"), IPRateLimitRPS: grpcIPRateLimitRPS, IPRateLimitBurst: grpcIPRateLimitBurst, + MaxInFlightPerIP: grpcMaxInFlightPerIP, RateLimitingEnabled: v.GetBool("grpc.rate-limiting-enabled"), TrustedProxyCIDRs: grpcTrustedProxyCIDRs, }, GRPCWeb: GRPCWebConfig{ - Enable: v.GetBool("grpc-web.enable"), - Address: v.GetString("grpc-web.address"), - EnableUnsafeCORS: v.GetBool("grpc-web.enable-unsafe-cors"), - MaxOpenConnections: grpcWebMaxOpenConnections, + Enable: v.GetBool("grpc-web.enable"), + Address: v.GetString("grpc-web.address"), + EnableUnsafeCORS: v.GetBool("grpc-web.enable-unsafe-cors"), + MaxOpenConnections: grpcWebMaxOpenConnections, + MaxConnectionsPerIP: grpcWebMaxConnectionsPerIP, }, StateSync: StateSyncConfig{ SnapshotInterval: v.GetUint64("state-sync.snapshot-interval"), diff --git a/sei-cosmos/server/config/config_fuzz_test.go b/sei-cosmos/server/config/config_fuzz_test.go index 101899a749..79ad666870 100644 --- a/sei-cosmos/server/config/config_fuzz_test.go +++ b/sei-cosmos/server/config/config_fuzz_test.go @@ -720,12 +720,14 @@ func TestGetConfigGRPCAbsentReads(t *testing.T) { }{ {"grpc.max-recv-msg-size", got.MaxRecvMsgSize, def.MaxRecvMsgSize}, {"grpc.max-open-connections", got.MaxOpenConnections, def.MaxOpenConnections}, + {"grpc.max-connections-per-ip", got.MaxConnectionsPerIP, def.MaxConnectionsPerIP}, {"grpc.max-connection-idle", got.MaxConnectionIdle, def.MaxConnectionIdle}, {"grpc.keepalive-time", got.KeepaliveTime, def.KeepaliveTime}, {"grpc.keepalive-timeout", got.KeepaliveTimeout, def.KeepaliveTimeout}, {"grpc.keepalive-min-time", got.KeepaliveMinTime, def.KeepaliveMinTime}, {"grpc.ip-rate-limit-rps", got.IPRateLimitRPS, def.IPRateLimitRPS}, {"grpc.ip-rate-limit-burst", got.IPRateLimitBurst, def.IPRateLimitBurst}, + {"grpc.max-in-flight-per-ip", got.MaxInFlightPerIP, def.MaxInFlightPerIP}, } { if c.absent != c.declared { t.Errorf("an absent %s resolved to %v rather than the declared %v, so its v.IsSet guard "+ diff --git a/sei-cosmos/server/config/toml.go b/sei-cosmos/server/config/toml.go index ead31582bf..75e1c66e2b 100644 --- a/sei-cosmos/server/config/toml.go +++ b/sei-cosmos/server/config/toml.go @@ -239,6 +239,12 @@ max-recv-msg-size = {{ .GRPC.MaxRecvMsgSize }} # MaxOpenConnections defines the maximum number of simultaneous open connections. 0 means unlimited. max-open-connections = {{ .GRPC.MaxOpenConnections }} +# max-connections-per-ip is the maximum number of simultaneous open connections one +# client address may hold, bounding its share of max-open-connections. Addresses are +# keyed the way rate-limit buckets are, so everything reaching the node over 127.0.0.1 +# shares one allowance, as does anything behind a single egress address. 0 means unlimited. +max-connections-per-ip = {{ .GRPC.MaxConnectionsPerIP }} + # MaxConnectionIdle is the duration after which an idle connection is closed (e.g. "5m"). 0 means infinity. max-connection-idle = "{{ .GRPC.MaxConnectionIdle }}" @@ -271,6 +277,15 @@ ip-rate-limit-rps = {{ .GRPC.IPRateLimitRPS }} # Zero disables per-IP throttling (same effect as ip-rate-limit-rps = 0). ip-rate-limit-burst = {{ .GRPC.IPRateLimitBurst }} +# max-in-flight-per-ip is the number of RPCs one client address may have in flight at +# once. An RPC is in flight from the moment its headers arrive until it ends, so this +# bounds concurrency where ip-rate-limit-rps bounds only arrival rate: without it a +# client can spend its bucket on headers alone, withhold the request bodies, and release +# them together. Excess requests are rejected with ResourceExhausted. +# It shares the same per-address keying as the buckets, and applies only when +# rate-limiting-enabled is true. Zero disables it. +max-in-flight-per-ip = {{ .GRPC.MaxInFlightPerIP }} + # rate-limiting-enabled is the master switch for gRPC rate-limit admission. # It governs gRPC-Web (:9091) as well as native gRPC (:9090), and both planes draw # from the same per-IP buckets. @@ -300,6 +315,10 @@ enable-unsafe-cors = {{ .GRPCWeb.EnableUnsafeCORS }} # MaxOpenConnections defines the number of maximum open connections. 0 means unlimited. max-open-connections = {{ .GRPCWeb.MaxOpenConnections }} +# max-connections-per-ip is the maximum number of simultaneous open connections one +# client address may hold, bounding its share of max-open-connections. 0 means unlimited. +max-connections-per-ip = {{ .GRPCWeb.MaxConnectionsPerIP }} + ############################################################################### ### Genesis Configuration (Auto-managed) ### ############################################################################### diff --git a/sei-cosmos/server/grpc/grpc_web.go b/sei-cosmos/server/grpc/grpc_web.go index 2b90504c19..812e8617fd 100644 --- a/sei-cosmos/server/grpc/grpc_web.go +++ b/sei-cosmos/server/grpc/grpc_web.go @@ -2,7 +2,6 @@ package grpc import ( "fmt" - "math" "net" "net/http" "time" @@ -46,12 +45,11 @@ func StartGRPCWeb(grpcSrv *grpc.Server, registry *ratelimiter.Registry, config c if err != nil { return nil, fmt.Errorf("[grpc-web] failed to listen on %s: %w", config.GRPCWeb.Address, err) } + // Same ordering as :9090: the per-IP cap sits below the global one, so an + // address at its limit cannot consume the shared budget to be refused. + listener = ratelimiter.ConnLimitListener(listener, ratelimiter.PlaneGRPCWeb, clampToMaxInt(config.GRPCWeb.MaxConnectionsPerIP)) if config.GRPCWeb.MaxOpenConnections > 0 { - maxConn := config.GRPCWeb.MaxOpenConnections - if maxConn > math.MaxInt { - maxConn = math.MaxInt - } - listener = netutil.LimitListener(listener, int(maxConn)) //nolint:gosec // G115: clamped to math.MaxInt above + listener = netutil.LimitListener(listener, clampToMaxInt(config.GRPCWeb.MaxOpenConnections)) } errCh := make(chan error, 1) diff --git a/sei-cosmos/server/grpc/inflight_test.go b/sei-cosmos/server/grpc/inflight_test.go new file mode 100644 index 0000000000..878e885c2b --- /dev/null +++ b/sei-cosmos/server/grpc/inflight_test.go @@ -0,0 +1,446 @@ +package grpc + +import ( + "context" + "io" + "net" + "net/http" + "sync" + "testing" + "time" + + gogogrpc "github.com/gogo/protobuf/grpc" + "github.com/stretchr/testify/require" + dbm "github.com/tendermint/tm-db" + "go.opentelemetry.io/otel/attribute" + sdkmetric "go.opentelemetry.io/otel/sdk/metric" + "go.opentelemetry.io/otel/sdk/metric/metricdata" + "google.golang.org/grpc" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/credentials/insecure" + "google.golang.org/grpc/status" + + "github.com/sei-protocol/sei-chain/ratelimiter" + "github.com/sei-protocol/sei-chain/sei-cosmos/baseapp" + "github.com/sei-protocol/sei-chain/sei-cosmos/client" + "github.com/sei-protocol/sei-chain/sei-cosmos/server/config" + "github.com/sei-protocol/sei-chain/sei-cosmos/testutil" + "github.com/sei-protocol/sei-chain/sei-cosmos/testutil/testdata" + sdk "github.com/sei-protocol/sei-chain/sei-cosmos/types" + moduletestutil "github.com/sei-protocol/sei-chain/sei-cosmos/types/module/testutil" +) + +// blockingQuery serves testdata.Query with an Echo that holds its slot until the +// test releases it or the client goes away. +// +// The concurrency cap is only observable while RPCs overlap, and an RPC overlaps +// another only for as long as its handler runs. +type blockingQuery struct { + testdata.QueryImpl + + entered chan struct{} + release chan struct{} + releaseOnce sync.Once +} + +func newBlockingQuery() *blockingQuery { + return &blockingQuery{ + entered: make(chan struct{}, 64), + release: make(chan struct{}), + } +} + +func (q *blockingQuery) Echo(ctx context.Context, req *testdata.EchoRequest) (*testdata.EchoResponse, error) { + q.entered <- struct{}{} + select { + case <-q.release: + return &testdata.EchoResponse{Message: req.Message}, nil + case <-ctx.Done(): + return nil, ctx.Err() + } +} + +// releaseAll lets every blocked handler return, now and in future. +func (q *blockingQuery) releaseAll() { q.releaseOnce.Do(func() { close(q.release) }) } + +// waitEntered blocks until n Echo handlers are running. +func (q *blockingQuery) waitEntered(t *testing.T, n int) { + t.Helper() + for range n { + select { + case <-q.entered: + case <-time.After(10 * time.Second): + t.Fatal("an Echo handler never started") + } + } +} + +// blockingQueryApp registers the blocking Query service directly on the gRPC +// server rather than on the BaseApp query router. +// +// The router dispatches through an ABCI query carrying an sdk.Context of its +// own, so a handler behind it never sees the client's context and could not +// observe a cancelled RPC. Registering the service directly keeps the real +// grpc-go lifecycle, which is what the release hook is being tested against. +type blockingQueryApp struct { + startGRPCServerApp + + query *blockingQuery +} + +func (a blockingQueryApp) RegisterGRPCServer(srv gogogrpc.Server) { + testdata.RegisterQueryServer(srv, a.query) +} + +// blockingServer is a live server whose Echo blocks, with the handles a test +// needs to drive it. +type blockingServer struct { + srv *grpc.Server + addr string + registry *ratelimiter.Registry + query *blockingQuery +} + +// startBlockingGRPCServer starts a real server through StartGRPCServer whose +// Echo blocks. It dials nothing, so a test counting connections counts its own. +func startBlockingGRPCServer(t *testing.T, cfg config.GRPCConfig) *blockingServer { + t.Helper() + + app := baseapp.NewBaseApp(t.Name(), dbm.NewMemDB(), nil, nil, &testutil.TestAppOpts{}) + app.MountStores(sdk.NewKVStoreKey("test")) + require.NoError(t, app.LoadLatestVersion()) + + encCfg := moduletestutil.MakeTestEncodingConfig() + testdata.RegisterInterfaces(encCfg.InterfaceRegistry) + app.SetInterfaceRegistry(encCfg.InterfaceRegistry) + + clientCtx := client.Context{}. + WithChainID("test-chain"). + WithTxConfig(encCfg.TxConfig). + WithInterfaceRegistry(encCfg.InterfaceRegistry) + + query := newBlockingQuery() + cfg.Address = freeTCPAddr(t) + srv, registry, err := StartGRPCServer(clientCtx, blockingQueryApp{startGRPCServerApp{app}, query}, cfg) + require.NoError(t, err) + t.Cleanup(srv.Stop) + t.Cleanup(query.releaseAll) + + return &blockingServer{srv: srv, addr: cfg.Address, registry: registry, query: query} +} + +// dial returns a client on a new connection to the server. +func (s *blockingServer) dial(t *testing.T) testdata.QueryClient { + t.Helper() + conn, err := grpc.Dial(s.addr, grpc.WithTransportCredentials(insecure.NewCredentials())) + require.NoError(t, err) + t.Cleanup(func() { _ = conn.Close() }) + return testdata.NewQueryClient(conn) +} + +// echoAsync starts an Echo and reports its error on a channel. +func (s *blockingServer) echoAsync(ctx context.Context, client testdata.QueryClient) <-chan error { + errCh := make(chan error, 1) + go func() { + _, err := client.Echo(ctx, &testdata.EchoRequest{Message: "hello"}) + errCh <- err + }() + return errCh +} + +// requireHeldSlots pins how many concurrency slots the loopback address holds. +func (s *blockingServer) requireHeldSlots(t *testing.T, want int) { + t.Helper() + require.Equal(t, want, s.registry.InFlightHeld("127.0.0.1")) +} + +// requireEventuallyNoHeldSlots waits for a release, which happens on the +// server's own goroutine after the client has already seen the RPC end. +func (s *blockingServer) requireEventuallyNoHeldSlots(t *testing.T) { + t.Helper() + require.Eventually(t, func() bool { + return s.registry.InFlightHeld("127.0.0.1") == 0 + }, 10*time.Second, 10*time.Millisecond, "a concurrency slot was never released") +} + +// inFlightLimitedGRPCConfig leaves the token bucket wide enough that only the +// concurrency cap can reject. +func inFlightLimitedGRPCConfig(maxInFlight int) config.GRPCConfig { + cfg := rateLimitedGRPCConfig(10_000, 10_000) + cfg.MaxInFlightPerIP = maxInFlight + return cfg +} + +func requireEchoErr(t *testing.T, errCh <-chan error) error { + t.Helper() + select { + case err := <-errCh: + return err + case <-time.After(10 * time.Second): + t.Fatal("an Echo never returned") + return nil + } +} + +// TestStartGRPCServer_InFlightCapRejectsBeyondLimit is the concurrency bound the +// token bucket does not provide: with tokens to spare, a third overlapping RPC +// from one address is refused while two are still running. +func TestStartGRPCServer_InFlightCapRejectsBeyondLimit(t *testing.T) { + reader := collectRejectionMetrics(t) + const method = "testdata.Query/Echo" + inFlightBefore := rejectionCounts(t, reader, inFlightRejectedMetric)[method] + rateBefore := rejectionCounts(t, reader, rateLimitRejectedMetric)[method] + + server := startBlockingGRPCServer(t, inFlightLimitedGRPCConfig(2)) + client := server.dial(t) + + first := server.echoAsync(t.Context(), client) + second := server.echoAsync(t.Context(), client) + server.query.waitEntered(t, 2) + + _, err := client.Echo(t.Context(), &testdata.EchoRequest{Message: "hello"}) + require.Equal(t, codes.ResourceExhausted, status.Code(err)) + + // The rejection is attributed to the concurrency limit, not the bucket, so + // an operator can tell the two controls apart. + require.Equal(t, inFlightBefore+1, rejectionCounts(t, reader, inFlightRejectedMetric)[method]) + require.Equal(t, rateBefore, rejectionCounts(t, reader, rateLimitRejectedMetric)[method]) + + server.query.releaseAll() + require.NoError(t, requireEchoErr(t, first)) + require.NoError(t, requireEchoErr(t, second)) +} + +// TestStartGRPCServer_InFlightSlotReleasedWhenRPCEnds pins the release hook: a +// slot held by a finished RPC is available to the next one. A slot that leaked +// would lock the address out for as long as the process ran. +func TestStartGRPCServer_InFlightSlotReleasedWhenRPCEnds(t *testing.T) { + server := startBlockingGRPCServer(t, inFlightLimitedGRPCConfig(1)) + client := server.dial(t) + + first := server.echoAsync(t.Context(), client) + server.query.waitEntered(t, 1) + server.requireHeldSlots(t, 1) + + _, err := client.Echo(t.Context(), &testdata.EchoRequest{Message: "hello"}) + require.Equal(t, codes.ResourceExhausted, status.Code(err)) + + server.query.releaseAll() + require.NoError(t, requireEchoErr(t, first)) + server.requireEventuallyNoHeldSlots(t) + + res, err := client.Echo(t.Context(), &testdata.EchoRequest{Message: "hello"}) + require.NoError(t, err) + require.Equal(t, "hello", res.Message) +} + +// TestStartGRPCServer_InFlightSlotReleasedOnClientCancel covers the termination +// path a served handler does not: the client abandons the RPC and the stream is +// reset. grpc-go emits the end-of-RPC event on that path too, which is why the +// release hangs off it rather than off the handler returning normally. +func TestStartGRPCServer_InFlightSlotReleasedOnClientCancel(t *testing.T) { + server := startBlockingGRPCServer(t, inFlightLimitedGRPCConfig(1)) + client := server.dial(t) + + ctx, cancel := context.WithCancel(t.Context()) + first := server.echoAsync(ctx, client) + server.query.waitEntered(t, 1) + server.requireHeldSlots(t, 1) + + cancel() + require.Error(t, requireEchoErr(t, first)) + server.requireEventuallyNoHeldSlots(t) +} + +// TestStartGRPCServer_UnknownMethodCannotLeakInFlightSlots is the leak the cap +// would otherwise carry. grpc-go answers an unknown or malformed method name +// without emitting the end-of-RPC event the release hangs off, so a slot taken +// for one would never come back and the address would end up locked out +// entirely — a worse outcome than the burst the cap prevents. +func TestStartGRPCServer_UnknownMethodCannotLeakInFlightSlots(t *testing.T) { + server := startBlockingGRPCServer(t, inFlightLimitedGRPCConfig(1)) + conn, err := grpc.Dial(server.addr, grpc.WithTransportCredentials(insecure.NewCredentials())) + require.NoError(t, err) + t.Cleanup(func() { _ = conn.Close() }) + + // Both shapes grpc-go answers before it emits anything: a name with no + // service segment, and a well-formed name for a service nobody registered. + for _, method := range []string{"/nope", "/nope.Service/Nope"} { + for range 10 { + err := conn.Invoke(t.Context(), method, &testdata.EchoRequest{}, &testdata.EchoResponse{}) + require.Equal(t, codes.Unimplemented, status.Code(err), "method %q", method) + } + } + + server.requireHeldSlots(t, 0) + + // The address still has its whole allowance, which is what a leak would have + // taken away. + server.query.releaseAll() + _, err = testdata.NewQueryClient(conn).Echo(t.Context(), &testdata.EchoRequest{Message: "hello"}) + require.NoError(t, err) +} + +// TestStartGRPCWeb_SharesInFlightSlotsWithGRPC pins that gRPC-Web draws on the +// same per-IP pool as :9090 rather than a second one that would double an +// address's concurrency, and that it gives its slot back. +func TestStartGRPCWeb_SharesInFlightSlotsWithGRPC(t *testing.T) { + server := startBlockingGRPCServer(t, inFlightLimitedGRPCConfig(1)) + require.NotNil(t, server.registry) + client := server.dial(t) + + webAddr := freeTCPAddr(t) + webSrv, err := StartGRPCWeb(server.srv, server.registry, config.Config{ + GRPCWeb: config.GRPCWebConfig{Enable: true, Address: webAddr}, + }) + require.NoError(t, err) + t.Cleanup(func() { _ = webSrv.Close() }) + + // The native plane holds the address's only slot. + first := server.echoAsync(t.Context(), client) + server.query.waitEntered(t, 1) + require.Equal(t, http.StatusTooManyRequests, postGRPCWeb(t, webAddr)) + + server.query.releaseAll() + require.NoError(t, requireEchoErr(t, first)) + server.requireEventuallyNoHeldSlots(t) + + // gRPC-Web hands its own slot back, so the pool is not drained by use. The + // stats handler must not release it a second time, which would mint a slot. + require.NotEqual(t, http.StatusTooManyRequests, postGRPCWeb(t, webAddr)) + server.requireEventuallyNoHeldSlots(t) +} + +// TestStartGRPCServer_InFlightCapDisabledAdmitsOverlappingRPCs pins the off +// switch: a zero limit leaves overlapping RPCs to the token bucket alone. +func TestStartGRPCServer_InFlightCapDisabledAdmitsOverlappingRPCs(t *testing.T) { + server := startBlockingGRPCServer(t, inFlightLimitedGRPCConfig(0)) + client := server.dial(t) + + running := make([]<-chan error, 0, 5) + for range 5 { + running = append(running, server.echoAsync(t.Context(), client)) + } + server.query.waitEntered(t, 5) + + server.query.releaseAll() + for _, errCh := range running { + require.NoError(t, requireEchoErr(t, errCh)) + } +} + +// TestStartGRPCServer_ConnectionsPerIPCapRefusesExcess bounds the layer below +// the RPC: the accepted sockets and HTTP/2 frame state one address can hold, +// which the per-RPC counter cannot see. +func TestStartGRPCServer_ConnectionsPerIPCapRefusesExcess(t *testing.T) { + cfg := inFlightLimitedGRPCConfig(0) + cfg.MaxConnectionsPerIP = 2 + server := startBlockingGRPCServer(t, cfg) + + held := dialRaw(t, server.addr) + requireConnAlive(t, dialRaw(t, server.addr)) + requireServerHungUp(t, dialRaw(t, server.addr)) + + // Closing a connection returns its slot. + require.NoError(t, held.Close()) + requireConnAlive(t, dialRaw(t, server.addr)) +} + +// TestStartGRPCServer_ConnectionsPerIPCapAppliesWithRateLimitingDisabled pins +// that the connection cap is independent of rate-limiting-enabled: it wraps the +// listener, which is below every control the rate-limit switch installs. +func TestStartGRPCServer_ConnectionsPerIPCapAppliesWithRateLimitingDisabled(t *testing.T) { + cfg := inFlightLimitedGRPCConfig(0) + cfg.RateLimitingEnabled = false + cfg.MaxConnectionsPerIP = 1 + server := startBlockingGRPCServer(t, cfg) + require.Nil(t, server.registry, "rate limiting is off, so there is no registry to admit against") + + requireConnAlive(t, dialRaw(t, server.addr)) + requireServerHungUp(t, dialRaw(t, server.addr)) +} + +// TestStartGRPCWeb_ConnectionsPerIPCapRefusesExcess covers the :9091 listener, +// which takes its cap from a config key of its own and enforces it with a +// counter of its own, so :9090 passing proves nothing about it. +func TestStartGRPCWeb_ConnectionsPerIPCapRefusesExcess(t *testing.T) { + reader := collectRejectionMetrics(t) + before := connRejectedCountsByPlane(t, reader) + + server := startBlockingGRPCServer(t, inFlightLimitedGRPCConfig(0)) + webAddr := freeTCPAddr(t) + webSrv, err := StartGRPCWeb(server.srv, server.registry, config.Config{ + GRPCWeb: config.GRPCWebConfig{ + Enable: true, + Address: webAddr, + MaxConnectionsPerIP: 2, + }, + }) + require.NoError(t, err) + t.Cleanup(func() { _ = webSrv.Close() }) + + held := dialRaw(t, webAddr) + requireConnAlive(t, dialRaw(t, webAddr)) + requireServerHungUp(t, dialRaw(t, webAddr)) + + // The refusal names the listener that made it, which is the only way an + // operator can tell a :9091 cap from the :9090 one. + after := connRejectedCountsByPlane(t, reader) + require.Equal(t, before[ratelimiter.PlaneGRPCWeb]+1, after[ratelimiter.PlaneGRPCWeb]) + require.Equal(t, before[ratelimiter.PlaneGRPC], after[ratelimiter.PlaneGRPC]) + + // Closing a connection returns its slot on this listener too. + require.NoError(t, held.Close()) + requireConnAlive(t, dialRaw(t, webAddr)) +} + +// connRejectedCountsByPlane returns the connection-rejection counter's value per +// plane label. The counter carries no method label, so rejectionCounts, which +// keys on method_namespace, cannot read it. +func connRejectedCountsByPlane(t *testing.T, reader *sdkmetric.ManualReader) map[string]int64 { + t.Helper() + var rm metricdata.ResourceMetrics + require.NoError(t, reader.Collect(t.Context(), &rm)) + + counts := map[string]int64{} + for _, sm := range rm.ScopeMetrics { + for _, m := range sm.Metrics { + if m.Name != connRejectedMetric { + continue + } + for _, dp := range m.Data.(metricdata.Sum[int64]).DataPoints { + if v, ok := dp.Attributes.Value(attribute.Key("plane")); ok { + counts[v.AsString()] += dp.Value + } + } + } + } + return counts +} + +func dialRaw(t *testing.T, addr string) net.Conn { + t.Helper() + conn, err := net.Dial("tcp", addr) + require.NoError(t, err) + t.Cleanup(func() { _ = conn.Close() }) + return conn +} + +// requireServerHungUp pins that the server closed conn rather than serving it, +// which is how an over-cap connection is refused. +func requireServerHungUp(t *testing.T, conn net.Conn) { + t.Helper() + require.NoError(t, conn.SetReadDeadline(time.Now().Add(10*time.Second))) + _, err := conn.Read(make([]byte, 1)) + require.ErrorIs(t, err, io.EOF) +} + +// requireConnAlive pins that the server kept conn: a connection under the cap is +// held open waiting for the client's HTTP/2 preface, so the read times out +// rather than reaching EOF. +func requireConnAlive(t *testing.T, conn net.Conn) { + t.Helper() + require.NoError(t, conn.SetReadDeadline(time.Now().Add(500*time.Millisecond))) + _, err := conn.Read(make([]byte, 1)) + require.NotErrorIs(t, err, io.EOF) +} diff --git a/sei-cosmos/server/grpc/rate_limit.go b/sei-cosmos/server/grpc/rate_limit.go index 8e70f52708..ffd5da8d73 100644 --- a/sei-cosmos/server/grpc/rate_limit.go +++ b/sei-cosmos/server/grpc/rate_limit.go @@ -6,17 +6,32 @@ import ( "google.golang.org/grpc" "google.golang.org/grpc/codes" + "google.golang.org/grpc/stats" "google.golang.org/grpc/status" "google.golang.org/grpc/tap" "github.com/sei-protocol/sei-chain/ratelimiter" ) -var errRateLimited = status.Error(codes.ResourceExhausted, "too many requests") +var ( + errRateLimited = status.Error(codes.ResourceExhausted, "too many requests") + errTooManyInFlight = status.Error(codes.ResourceExhausted, "too many concurrent requests") +) // admittedIPKey addresses the client IP recorded when the tap handler charges an RPC. type admittedIPKey struct{} +// inFlightIPKey addresses the client IP a concurrency slot was taken for, and +// marks the RPC as one InFlightStatsHandler must release. +type inFlightIPKey struct{} + +// inFlightIP returns the client IP a concurrency slot is held for on this RPC +// and whether one is held at all. +func inFlightIP(ctx context.Context) (string, bool) { + ip, ok := ctx.Value(inFlightIPKey{}).(string) + return ip, ok +} + // admittedIP returns the client IP the tap handler charged for this RPC and // whether the tap handler ran at all. func admittedIP(ctx context.Context) (string, bool) { @@ -41,7 +56,71 @@ func RateLimitTapHandle(registry *ratelimiter.Registry) tap.ServerInHandle { if !registry.Allow(ctx, ip, ratelimiter.PlaneGRPC, info.FullMethodName) { return ctx, errRateLimited } - return context.WithValue(ctx, admittedIPKey{}, ip), nil + ctx = context.WithValue(ctx, admittedIPKey{}, ip) + return acquireInFlightSlot(ctx, registry, ip, info.FullMethodName) + } +} + +// acquireInFlightSlot takes ip's concurrency slot for a native-gRPC RPC and +// returns the context InFlightStatsHandler releases it from, or an error when +// the IP already holds its share. +// +// The slot is taken only for a method the server actually serves, and that is a +// correctness requirement rather than an optimisation. grpc-go returns from +// handleStream for an unknown or malformed method name without ever emitting +// the stats events the release hangs off, so a client spraying HEADERS at +// "/nope" would leak slots until the IP could open nothing at all — a worse +// outcome than the burst the cap exists to prevent. An unknown method is +// answered with Unimplemented either way, and the token bucket still charges it. +// +// It follows that the cap is inert until Registry.SetKnownGRPCMethods has run, +// which StartGRPCServer does before it serves. +func acquireInFlightSlot(ctx context.Context, registry *ratelimiter.Registry, ip, fullMethod string) (context.Context, error) { + if !registry.IsKnownGRPCMethod(fullMethod) { + return ctx, nil + } + if !registry.AcquireInFlight(ctx, ip, ratelimiter.PlaneGRPC, fullMethod) { + return ctx, errTooManyInFlight + } + return context.WithValue(ctx, inFlightIPKey{}, ip), nil +} + +// InFlightStatsHandler returns the stats handler that releases the concurrency +// slots RateLimitTapHandle takes. registry must be non-nil. +// +// stats.End is the only hook that brackets a stream the tap admitted, whatever +// ends it: a served handler, a client reset, a lost connection, or a panic +// unwinding through grpc-go's own deferred emit. An interceptor cannot, because +// grpc-go reaches one only after decoding the request message, which is +// precisely the event a stockpiling client withholds. +func InFlightStatsHandler(registry *ratelimiter.Registry) stats.Handler { + return inFlightStatsHandler{registry: registry} +} + +type inFlightStatsHandler struct { + registry *ratelimiter.Registry +} + +func (inFlightStatsHandler) TagRPC(ctx context.Context, _ *stats.RPCTagInfo) context.Context { + return ctx +} +func (inFlightStatsHandler) TagConn(ctx context.Context, _ *stats.ConnTagInfo) context.Context { + return ctx +} +func (inFlightStatsHandler) HandleConn(context.Context, stats.ConnStats) {} + +// HandleRPC releases the slot this RPC holds once, when the RPC ends. +// +// Only an RPC whose context carries the marker holds one. gRPC-Web reaches the +// same server through ServeHTTP, which emits the same stats events but never +// runs the tap handler; RateLimitHTTPMiddleware releases that plane's slot under +// its own defer, and leaves no marker here, so the slot is not returned twice. +func (h inFlightStatsHandler) HandleRPC(ctx context.Context, rpcStats stats.RPCStats) { + if _, ok := rpcStats.(*stats.End); !ok { + return + } + if ip, held := inFlightIP(ctx); held { + h.registry.ReleaseInFlight(ip) } } @@ -66,6 +145,14 @@ func RateLimitHTTPMiddleware(registry *ratelimiter.Registry, next http.Handler) http.Error(w, "too many requests", http.StatusTooManyRequests) return } + // The slot is released here rather than from the stats handler, because + // this handler brackets the whole request already. No marker goes on the + // context, so the stats handler does not release it a second time. + if !registry.AcquireInFlight(r.Context(), ip, ratelimiter.PlaneGRPC, r.URL.Path) { + http.Error(w, "too many concurrent requests", http.StatusTooManyRequests) + return + } + defer registry.ReleaseInFlight(ip) next.ServeHTTP(w, r.WithContext(context.WithValue(r.Context(), admittedIPKey{}, ip))) }) } diff --git a/sei-cosmos/server/grpc/server.go b/sei-cosmos/server/grpc/server.go index 668177941f..ec3141d699 100644 --- a/sei-cosmos/server/grpc/server.go +++ b/sei-cosmos/server/grpc/server.go @@ -44,7 +44,7 @@ func rateLimitServerOptions(cfg config.GRPCConfig) ([]grpc.ServerOption, *rateli "ip-rate-limit-burst", cfg.IPRateLimitBurst, ) } - return []grpc.ServerOption{ + opts := []grpc.ServerOption{ // Admission happens before the request is decoded: the tap handler // covers native gRPC, and RateLimitHTTPMiddleware covers gRPC-Web, // which reaches this server through ServeHTTP. The interceptors charge @@ -53,7 +53,29 @@ func rateLimitServerOptions(cfg config.GRPCConfig) ([]grpc.ServerOption, *rateli grpc.InTapHandle(RateLimitTapHandle(registry)), grpc.ChainUnaryInterceptor(UnaryRateLimitInterceptor(registry)), grpc.ChainStreamInterceptor(StreamRateLimitInterceptor(registry)), - }, registry, nil + } + // The tap handler takes a per-IP concurrency slot alongside the token, and + // the stats handler is what gives it back, on every path a stream can end. + // Registering one at all makes grpc-go build and dispatch a stats event for + // every phase of every RPC, so with the cap off it is left out rather than + // left inert. + if cfg.MaxInFlightPerIP > 0 { + opts = append(opts, grpc.StatsHandler(InFlightStatsHandler(registry))) + } + return opts, registry, nil +} + +// clampToMaxInt returns n as an int, saturating rather than wrapping. +// +// The connection limits are configured as unsigned and consumed as signed, and a +// value above the signed maximum would otherwise wrap to a negative one, which +// both listeners read as "no limit" — the opposite of what an operator writing a +// very large number asked for. +func clampToMaxInt(n uint) int { + if n > math.MaxInt { + return math.MaxInt + } + return int(n) //nolint:gosec // G115: clamped to math.MaxInt above } // registeredMethods returns the "service/Method" name of every method served by @@ -136,12 +158,12 @@ func StartGRPCServer(clientCtx client.Context, app types.Application, cfg config if err != nil { return nil, nil, err } + // The per-IP cap wraps the raw listener and the global cap wraps that, so a + // connection one address is refused never occupies a slot in the global + // budget it would otherwise be able to hold whole. + listener = ratelimiter.ConnLimitListener(listener, ratelimiter.PlaneGRPC, clampToMaxInt(cfg.MaxConnectionsPerIP)) if cfg.MaxOpenConnections > 0 { - maxConn := cfg.MaxOpenConnections - if maxConn > math.MaxInt { - maxConn = math.MaxInt - } - listener = netutil.LimitListener(listener, int(maxConn)) //nolint:gosec // G115: clamped to math.MaxInt above + listener = netutil.LimitListener(listener, clampToMaxInt(cfg.MaxOpenConnections)) } errCh := make(chan error) diff --git a/sei-cosmos/server/grpc/server_rate_limit_test.go b/sei-cosmos/server/grpc/server_rate_limit_test.go index c62970522a..575141f999 100644 --- a/sei-cosmos/server/grpc/server_rate_limit_test.go +++ b/sei-cosmos/server/grpc/server_rate_limit_test.go @@ -5,6 +5,7 @@ import ( "context" "net" "net/http" + "sync" "testing" "github.com/stretchr/testify/require" @@ -120,34 +121,66 @@ func TestStartGRPCServer_RateLimitingEnabledInstallsInterceptors(t *testing.T) { require.Contains(t, rejectionMethodLabels(t, reader), "testdata.Query/Echo") } +// rateLimitRejectedMetric and inFlightRejectedMetric are the two rejection +// counters, kept apart so a test can say which control did the rejecting. +const ( + rateLimitRejectedMetric = "rpc_rate_limit_rejected_total" + inFlightRejectedMetric = "rpc_inflight_rejected_total" + connRejectedMetric = "rpc_connection_rejected_total" +) + +var ( + rejectionReaderOnce sync.Once + rejectionReader *sdkmetric.ManualReader +) + +// collectRejectionMetrics returns the reader the rejection counters record into. +// +// One reader for the whole package rather than one per test, because the global +// meter provider takes only the first provider set: the counters are created at +// package init and delegate to that one, so a second reader would collect +// nothing and every assertion on it would pass vacuously. The counters are +// therefore cumulative across tests, so assert on what a test added rather than +// on a total. func collectRejectionMetrics(t *testing.T) *sdkmetric.ManualReader { t.Helper() - reader := sdkmetric.NewManualReader() - prev := otel.GetMeterProvider() - otel.SetMeterProvider(sdkmetric.NewMeterProvider(sdkmetric.WithReader(reader))) - t.Cleanup(func() { otel.SetMeterProvider(prev) }) - return reader + rejectionReaderOnce.Do(func() { + rejectionReader = sdkmetric.NewManualReader() + otel.SetMeterProvider(sdkmetric.NewMeterProvider(sdkmetric.WithReader(rejectionReader))) + }) + return rejectionReader } func rejectionMethodLabels(t *testing.T, reader *sdkmetric.ManualReader) []string { + t.Helper() + labels := []string{} + for label := range rejectionCounts(t, reader, rateLimitRejectedMetric) { + labels = append(labels, label) + } + return labels +} + +// rejectionCounts returns the named rejection counter's value per +// method_namespace label. +func rejectionCounts(t *testing.T, reader *sdkmetric.ManualReader, name string) map[string]int64 { t.Helper() var rm metricdata.ResourceMetrics require.NoError(t, reader.Collect(context.Background(), &rm)) - labels := []string{} + counts := map[string]int64{} for _, sm := range rm.ScopeMetrics { for _, m := range sm.Metrics { - if m.Name != "rpc_rate_limit_rejected_total" { + if m.Name != name { continue } for _, dp := range m.Data.(metricdata.Sum[int64]).DataPoints { if v, ok := dp.Attributes.Value(attribute.Key("method_namespace")); ok { - labels = append(labels, v.AsString()) + counts[v.AsString()] += dp.Value } } } } - return labels + return counts } // TestStartGRPCServer_RateLimitingDisabledSkipsInterceptors pins the master