From 4ef28e00e5b69ad0a492a04d6bc748152611662b Mon Sep 17 00:00:00 2001 From: Amir Deris Date: Wed, 2 Sep 2026 14:32:08 +0200 Subject: [PATCH 1/4] PLT-1125: Bound concurrent in-flight RPCs and connections per IP on gRPC Per-IP admission on the gRPC query plane charges one token at the HTTP/2 HEADERS frame and records the admitted IP on the RPC context. A client can spend a token on headers alone, withhold the request body while the bucket refills, repeat across streams, and release every body at once. The token bucket smooths arrival rate; it caps no concurrency. Two orthogonal controls close that, each useful on its own. Per-IP connection cap. ConnLimitListener bounds simultaneously-open connections per client address on :9090 and :9091, wrapping the raw listener inside the global cap so a refused connection never spends a global slot. Over-cap connections are closed rather than surfaced as an Accept error, which would turn one client's excess into an outage for everyone else. It applies whether or not rate limiting is enabled. Per-IP in-flight RPC cap. RateLimitTapHandle takes a slot at the HEADERS frame, after the token check, and InFlightStatsHandler returns it on stats.End -- the only hook that brackets a stream the tap admitted, whatever ends it. An interceptor cannot: grpc-go reaches one only after decoding the request message, which is precisely the event a stockpiling client withholds. Two failure modes shape the implementation. grpc-go answers an unknown or malformed method name without emitting stats at all, so acquisition is gated on the registered method set: a leaked slot fails closed and locks an IP out permanently, which is worse than the burst. And gRPC-Web reaches the same handleStream through ServeHTTP and emits the same stats events without ever running the tap, so RateLimitHTTPMiddleware releases its own slot under a defer and leaves no context marker, keeping the stats handler from returning it twice. New keys, all read behind presence guards: [grpc] max-connections-per-ip, [grpc-web] max-connections-per-ip, and [grpc] max-in-flight-per-ip, each defaulting to 100. Rejections are counted by rpc_inflight_rejected_total and rpc_connection_rejected_total, siblings of rpc_rate_limit_rejected_total so an operator can tell the controls apart. collectRejectionMetrics now shares one reader across the package. It installed a fresh meter provider per test, but the otel global takes only the first, so a second caller left the earlier test collecting nothing. Co-Authored-By: Claude Opus 5 (1M context) --- CHANGELOG.md | 2 + config/cosmosbase/agreement_test.go | 2 + config/cosmosbase/cosmosbase.go | 4 +- config/cosmosbase/cosmosbase_test.go | 5 +- ratelimiter/conn_limit.go | 83 ++++ ratelimiter/conn_limit_test.go | 118 ++++++ ratelimiter/inflight.go | 110 ++++++ ratelimiter/inflight_test.go | 118 ++++++ ratelimiter/metrics.go | 14 +- ratelimiter/registry.go | 17 +- sei-cosmos/server/config/config.go | 73 +++- sei-cosmos/server/config/config_fuzz_test.go | 2 + sei-cosmos/server/config/toml.go | 19 + sei-cosmos/server/grpc/grpc_web.go | 10 +- sei-cosmos/server/grpc/inflight_test.go | 371 ++++++++++++++++++ sei-cosmos/server/grpc/rate_limit.go | 91 ++++- sei-cosmos/server/grpc/server.go | 26 +- .../server/grpc/server_rate_limit_test.go | 50 ++- 18 files changed, 1080 insertions(+), 35 deletions(-) create mode 100644 ratelimiter/conn_limit.go create mode 100644 ratelimiter/conn_limit_test.go create mode 100644 ratelimiter/inflight.go create mode 100644 ratelimiter/inflight_test.go create mode 100644 sei-cosmos/server/grpc/inflight_test.go diff --git a/CHANGELOG.md b/CHANGELOG.md index 6b29f730e2..30e1ae0e24 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}`. +* [#4067](https://github.com/sei-protocol/sei-chain/pull/4067) 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 100 each) cap one address's share of the global connection budget on :9090 and :9091, and are on 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, which closes the gap where a client could spend its token bucket on headers alone, withhold the request bodies, and release them together. 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. +* [#4067](https://github.com/sei-protocol/sei-chain/pull/4067) **Connections from a single IP are now capped on :9090 and :9091.** `max-connections-per-ip` defaults to 100 on both planes and applies whether or not rate limiting is enabled, so an address already holding 100 connections has further ones accepted and immediately closed. Keying matches the rate-limit buckets: everything reaching the node over `127.0.0.1` shares one allowance, as does any population behind a single egress address that `trusted-proxy-cidrs` does not cover, and IPv6 clients share per /64. **Operators running a co-located indexer, a shared-NAT client population, or an ingress tier that is not listed in `trusted-proxy-cidrs` should raise `[grpc] max-connections-per-ip` and `[grpc-web] max-connections-per-ip`, or set them to `0` for the previous unlimited behaviour, before upgrading.** The companion `[grpc] max-in-flight-per-ip` (default 100 concurrent RPCs per address) only takes effect when `rate-limiting-enabled = true`, and starves shared-egress clients in the same way; 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..cb4abec081 100644 --- a/config/cosmosbase/cosmosbase.go +++ b/config/cosmosbase/cosmosbase.go @@ -96,8 +96,8 @@ 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 +// Eleven of these seventeen 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. func grpcDefaults(mode registry.Mode) any { return forMode(mode).GRPC } 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..4fa5e8841b --- /dev/null +++ b/ratelimiter/conn_limit.go @@ -0,0 +1,83 @@ +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 the way rate-limit buckets are, so a client rotating +// within an IPv6 /64 does not get a fresh allowance per address, and every +// client sharing an address shares one allowance. +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 + } + 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/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..8d50c9e4fb 100644 --- a/sei-cosmos/server/config/config.go +++ b/sei-cosmos/server/config/config.go @@ -32,10 +32,22 @@ 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 = 100 + // 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. It is a tenth of the global budget, so no single address can crowd + // the rest out of it, and it is well above what any ordinary client opens. + // 0 means unlimited. + DefaultGRPCMaxConnectionsPerIP = 100 + // 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 +95,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 +263,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 +307,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 +331,7 @@ func (c GRPCConfig) RateLimiterConfig() ratelimiter.Config { RPS: c.IPRateLimitRPS, Burst: c.IPRateLimitBurst, TrustedProxyCIDRs: c.TrustedProxyCIDRs, + MaxInFlightPerIP: c.MaxInFlightPerIP, } } @@ -311,6 +348,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 +458,7 @@ func DefaultConfig() *Config { Address: DefaultGRPCAddress, MaxRecvMsgSize: DefaultGRPCMaxRecvMsgSize, MaxOpenConnections: DefaultGRPCMaxOpenConnections, + MaxConnectionsPerIP: DefaultGRPCMaxConnectionsPerIP, MaxConnectionIdle: DefaultGRPCMaxConnectionIdle, MaxConnectionAge: DefaultGRPCMaxConnectionAge, MaxConnectionAgeGrace: DefaultGRPCMaxConnectionAgeGrace, @@ -426,6 +468,7 @@ func DefaultConfig() *Config { KeepalivePermitWithoutStream: DefaultGRPCKeepalivePermitWithoutStream, IPRateLimitRPS: DefaultGRPCIPRateLimitRPS, IPRateLimitBurst: DefaultGRPCIPRateLimitBurst, + MaxInFlightPerIP: DefaultGRPCMaxInFlightPerIP, RateLimitingEnabled: false, TrustedProxyCIDRs: nil, }, @@ -438,9 +481,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,6 +625,10 @@ 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 @@ -593,6 +641,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 +677,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 +734,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 +744,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..dc266efe08 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.PlaneGRPC, 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..0d17800740 --- /dev/null +++ b/sei-cosmos/server/grpc/inflight_test.go @@ -0,0 +1,371 @@ +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" + "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(context.Background(), client) + second := server.echoAsync(context.Background(), client) + server.query.waitEntered(t, 2) + + _, err := client.Echo(context.Background(), &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(context.Background(), client) + server.query.waitEntered(t, 1) + server.requireHeldSlots(t, 1) + + _, err := client.Echo(context.Background(), &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(context.Background(), &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(context.Background()) + 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(context.Background(), 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(context.Background(), &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(context.Background(), 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(context.Background(), 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)) +} + +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..6e253451e6 100644 --- a/sei-cosmos/server/grpc/server.go +++ b/sei-cosmos/server/grpc/server.go @@ -51,11 +51,27 @@ func rateLimitServerOptions(cfg config.GRPCConfig) ([]grpc.ServerOption, *rateli // what neither can see, each message on an established stream, and // admit any RPC that arrived uncharged. grpc.InTapHandle(RateLimitTapHandle(registry)), + // The tap handler takes a per-IP concurrency slot alongside the token; + // this is what gives it back, on every path a stream can end. + grpc.StatsHandler(InFlightStatsHandler(registry)), grpc.ChainUnaryInterceptor(UnaryRateLimitInterceptor(registry)), grpc.ChainStreamInterceptor(StreamRateLimitInterceptor(registry)), }, 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 // srv. func registeredMethods(srv *grpc.Server) []string { @@ -136,12 +152,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..cfda06990d 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,65 @@ 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" +) + +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 From b1b4d2e6504eb894c44574573b5d27d51b6acc4a Mon Sep 17 00:00:00 2001 From: Amir Deris Date: Wed, 2 Sep 2026 14:33:15 +0200 Subject: [PATCH 2/4] Point the CHANGELOG entries at the PR they landed in Co-Authored-By: Claude Opus 5 (1M context) --- CHANGELOG.md | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 30e1ae0e24..6d576e6094 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -37,7 +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}`. -* [#4067](https://github.com/sei-protocol/sei-chain/pull/4067) 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 100 each) cap one address's share of the global connection budget on :9090 and :9091, and are on 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, which closes the gap where a client could spend its token bucket on headers alone, withhold the request bodies, and release them together. 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}`. +* [#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 100 each) cap one address's share of the global connection budget on :9090 and :9091, and are on 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, which closes the gap where a client could spend its token bucket on headers alone, withhold the request bodies, and release them together. 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. @@ -49,7 +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. -* [#4067](https://github.com/sei-protocol/sei-chain/pull/4067) **Connections from a single IP are now capped on :9090 and :9091.** `max-connections-per-ip` defaults to 100 on both planes and applies whether or not rate limiting is enabled, so an address already holding 100 connections has further ones accepted and immediately closed. Keying matches the rate-limit buckets: everything reaching the node over `127.0.0.1` shares one allowance, as does any population behind a single egress address that `trusted-proxy-cidrs` does not cover, and IPv6 clients share per /64. **Operators running a co-located indexer, a shared-NAT client population, or an ingress tier that is not listed in `trusted-proxy-cidrs` should raise `[grpc] max-connections-per-ip` and `[grpc-web] max-connections-per-ip`, or set them to `0` for the previous unlimited behaviour, before upgrading.** The companion `[grpc] max-in-flight-per-ip` (default 100 concurrent RPCs per address) only takes effect when `rate-limiting-enabled = true`, and starves shared-egress clients in the same way; size it against your heaviest client's concurrency. +* [#4078](https://github.com/sei-protocol/sei-chain/pull/4078) **Connections from a single IP are now capped on :9090 and :9091.** `max-connections-per-ip` defaults to 100 on both planes and applies whether or not rate limiting is enabled, so an address already holding 100 connections has further ones accepted and immediately closed. Keying matches the rate-limit buckets: everything reaching the node over `127.0.0.1` shares one allowance, as does any population behind a single egress address that `trusted-proxy-cidrs` does not cover, and IPv6 clients share per /64. **Operators running a co-located indexer, a shared-NAT client population, or an ingress tier that is not listed in `trusted-proxy-cidrs` should raise `[grpc] max-connections-per-ip` and `[grpc-web] max-connections-per-ip`, or set them to `0` for the previous unlimited behaviour, before upgrading.** The companion `[grpc] max-in-flight-per-ip` (default 100 concurrent RPCs per address) only takes effect when `rate-limiting-enabled = true`, and starves shared-egress clients in the same way; 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`. From e6fcbf896da827a9b8dda52c1d45b6bb490c393a Mon Sep 17 00:00:00 2001 From: Amir Deris Date: Wed, 2 Sep 2026 15:50:48 +0200 Subject: [PATCH 3/4] Default max-connections-per-ip to 0 (unlimited) Preserve pre-upgrade behavior unless an operator opts in to a per-IP connection cap. Update CHANGELOG to match. Co-authored-by: Cursor --- CHANGELOG.md | 4 ++-- sei-cosmos/server/config/config.go | 12 +++++------- 2 files changed, 7 insertions(+), 9 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 6d576e6094..a963526a5f 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -37,7 +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 100 each) cap one address's share of the global connection budget on :9090 and :9091, and are on 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, which closes the gap where a client could spend its token bucket on headers alone, withhold the request bodies, and release them together. 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}`. +* [#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. @@ -49,7 +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) **Connections from a single IP are now capped on :9090 and :9091.** `max-connections-per-ip` defaults to 100 on both planes and applies whether or not rate limiting is enabled, so an address already holding 100 connections has further ones accepted and immediately closed. Keying matches the rate-limit buckets: everything reaching the node over `127.0.0.1` shares one allowance, as does any population behind a single egress address that `trusted-proxy-cidrs` does not cover, and IPv6 clients share per /64. **Operators running a co-located indexer, a shared-NAT client population, or an ingress tier that is not listed in `trusted-proxy-cidrs` should raise `[grpc] max-connections-per-ip` and `[grpc-web] max-connections-per-ip`, or set them to `0` for the previous unlimited behaviour, before upgrading.** The companion `[grpc] max-in-flight-per-ip` (default 100 concurrent RPCs per address) only takes effect when `rate-limiting-enabled = true`, and starves shared-egress clients in the same way; size it against your heaviest client's concurrency. +* [#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. Keying matches the rate-limit buckets: everything reaching the node over `127.0.0.1` shares one allowance, as does any population behind a single egress address that `trusted-proxy-cidrs` does not cover, and IPv6 clients share per /64. **Operators exposing public gRPC query endpoints 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).** 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/sei-cosmos/server/config/config.go b/sei-cosmos/server/config/config.go index 8d50c9e4fb..70db7578e3 100644 --- a/sei-cosmos/server/config/config.go +++ b/sei-cosmos/server/config/config.go @@ -35,7 +35,7 @@ const ( // DefaultGRPCWebMaxConnectionsPerIP defines the default maximum number of // simultaneous open connections one client address may hold on the gRPC-web // server. 0 means unlimited. - DefaultGRPCWebMaxConnectionsPerIP = 100 + DefaultGRPCWebMaxConnectionsPerIP = 0 // DefaultGRPCMaxOpenConnections defines the default maximum number of // simultaneous open connections for the gRPC server. 0 means unlimited. @@ -43,10 +43,8 @@ const ( // DefaultGRPCMaxConnectionsPerIP defines the default maximum number of // simultaneous open connections one client address may hold on the gRPC - // server. It is a tenth of the global budget, so no single address can crowd - // the rest out of it, and it is well above what any ordinary client opens. - // 0 means unlimited. - DefaultGRPCMaxConnectionsPerIP = 100 + // 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. @@ -631,8 +629,8 @@ func GetConfig(v *viper.Viper) (Config, error) { } // 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") From 41a7c6a6c9c148c6f259a152288c3c53226d3179 Mon Sep 17 00:00:00 2001 From: Amir Deris Date: Thu, 3 Sep 2026 16:19:31 +0200 Subject: [PATCH 4/4] Addressing AI feedback --- CHANGELOG.md | 2 +- config/cosmosbase/cosmosbase.go | 9 +- ratelimiter/conn_limit.go | 11 ++- ratelimiter/method_bucket.go | 5 + sei-cosmos/server/grpc/grpc_web.go | 2 +- sei-cosmos/server/grpc/inflight_test.go | 97 ++++++++++++++++--- sei-cosmos/server/grpc/server.go | 16 ++- .../server/grpc/server_rate_limit_test.go | 1 + 8 files changed, 118 insertions(+), 25 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index a963526a5f..8c10fb26cf 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -49,7 +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. Keying matches the rate-limit buckets: everything reaching the node over `127.0.0.1` shares one allowance, as does any population behind a single egress address that `trusted-proxy-cidrs` does not cover, and IPv6 clients share per /64. **Operators exposing public gRPC query endpoints 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).** 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. +* [#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/cosmosbase.go b/config/cosmosbase/cosmosbase.go index cb4abec081..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. // -// Eleven of these seventeen 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/ratelimiter/conn_limit.go b/ratelimiter/conn_limit.go index 4fa5e8841b..f9fd7619bd 100644 --- a/ratelimiter/conn_limit.go +++ b/ratelimiter/conn_limit.go @@ -18,9 +18,11 @@ import ( // 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 the way rate-limit buckets are, so a client rotating -// within an IPv6 /64 does not get a fresh allowance per address, and every -// client sharing an address shares one allowance. +// 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 @@ -52,6 +54,9 @@ func (l *connLimitListener) Accept() (net.Conn, error) { 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 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/sei-cosmos/server/grpc/grpc_web.go b/sei-cosmos/server/grpc/grpc_web.go index dc266efe08..812e8617fd 100644 --- a/sei-cosmos/server/grpc/grpc_web.go +++ b/sei-cosmos/server/grpc/grpc_web.go @@ -47,7 +47,7 @@ func StartGRPCWeb(grpcSrv *grpc.Server, registry *ratelimiter.Registry, config c } // 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.PlaneGRPC, clampToMaxInt(config.GRPCWeb.MaxConnectionsPerIP)) + listener = ratelimiter.ConnLimitListener(listener, ratelimiter.PlaneGRPCWeb, clampToMaxInt(config.GRPCWeb.MaxConnectionsPerIP)) if config.GRPCWeb.MaxOpenConnections > 0 { listener = netutil.LimitListener(listener, clampToMaxInt(config.GRPCWeb.MaxOpenConnections)) } diff --git a/sei-cosmos/server/grpc/inflight_test.go b/sei-cosmos/server/grpc/inflight_test.go index 0d17800740..878e885c2b 100644 --- a/sei-cosmos/server/grpc/inflight_test.go +++ b/sei-cosmos/server/grpc/inflight_test.go @@ -12,6 +12,9 @@ import ( 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" @@ -191,11 +194,11 @@ func TestStartGRPCServer_InFlightCapRejectsBeyondLimit(t *testing.T) { server := startBlockingGRPCServer(t, inFlightLimitedGRPCConfig(2)) client := server.dial(t) - first := server.echoAsync(context.Background(), client) - second := server.echoAsync(context.Background(), client) + first := server.echoAsync(t.Context(), client) + second := server.echoAsync(t.Context(), client) server.query.waitEntered(t, 2) - _, err := client.Echo(context.Background(), &testdata.EchoRequest{Message: "hello"}) + _, 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 @@ -215,18 +218,18 @@ func TestStartGRPCServer_InFlightSlotReleasedWhenRPCEnds(t *testing.T) { server := startBlockingGRPCServer(t, inFlightLimitedGRPCConfig(1)) client := server.dial(t) - first := server.echoAsync(context.Background(), client) + first := server.echoAsync(t.Context(), client) server.query.waitEntered(t, 1) server.requireHeldSlots(t, 1) - _, err := client.Echo(context.Background(), &testdata.EchoRequest{Message: "hello"}) + _, 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(context.Background(), &testdata.EchoRequest{Message: "hello"}) + res, err := client.Echo(t.Context(), &testdata.EchoRequest{Message: "hello"}) require.NoError(t, err) require.Equal(t, "hello", res.Message) } @@ -239,7 +242,7 @@ func TestStartGRPCServer_InFlightSlotReleasedOnClientCancel(t *testing.T) { server := startBlockingGRPCServer(t, inFlightLimitedGRPCConfig(1)) client := server.dial(t) - ctx, cancel := context.WithCancel(context.Background()) + ctx, cancel := context.WithCancel(t.Context()) first := server.echoAsync(ctx, client) server.query.waitEntered(t, 1) server.requireHeldSlots(t, 1) @@ -264,7 +267,7 @@ func TestStartGRPCServer_UnknownMethodCannotLeakInFlightSlots(t *testing.T) { // 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(context.Background(), method, &testdata.EchoRequest{}, &testdata.EchoResponse{}) + err := conn.Invoke(t.Context(), method, &testdata.EchoRequest{}, &testdata.EchoResponse{}) require.Equal(t, codes.Unimplemented, status.Code(err), "method %q", method) } } @@ -274,7 +277,7 @@ func TestStartGRPCServer_UnknownMethodCannotLeakInFlightSlots(t *testing.T) { // The address still has its whole allowance, which is what a leak would have // taken away. server.query.releaseAll() - _, err = testdata.NewQueryClient(conn).Echo(context.Background(), &testdata.EchoRequest{Message: "hello"}) + _, err = testdata.NewQueryClient(conn).Echo(t.Context(), &testdata.EchoRequest{Message: "hello"}) require.NoError(t, err) } @@ -294,7 +297,7 @@ func TestStartGRPCWeb_SharesInFlightSlotsWithGRPC(t *testing.T) { t.Cleanup(func() { _ = webSrv.Close() }) // The native plane holds the address's only slot. - first := server.echoAsync(context.Background(), client) + first := server.echoAsync(t.Context(), client) server.query.waitEntered(t, 1) require.Equal(t, http.StatusTooManyRequests, postGRPCWeb(t, webAddr)) @@ -316,7 +319,7 @@ func TestStartGRPCServer_InFlightCapDisabledAdmitsOverlappingRPCs(t *testing.T) running := make([]<-chan error, 0, 5) for range 5 { - running = append(running, server.echoAsync(context.Background(), client)) + running = append(running, server.echoAsync(t.Context(), client)) } server.query.waitEntered(t, 5) @@ -343,6 +346,78 @@ func TestStartGRPCServer_ConnectionsPerIPCapRefusesExcess(t *testing.T) { 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) diff --git a/sei-cosmos/server/grpc/server.go b/sei-cosmos/server/grpc/server.go index 6e253451e6..ec3141d699 100644 --- a/sei-cosmos/server/grpc/server.go +++ b/sei-cosmos/server/grpc/server.go @@ -44,19 +44,25 @@ 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 // what neither can see, each message on an established stream, and // admit any RPC that arrived uncharged. grpc.InTapHandle(RateLimitTapHandle(registry)), - // The tap handler takes a per-IP concurrency slot alongside the token; - // this is what gives it back, on every path a stream can end. - grpc.StatsHandler(InFlightStatsHandler(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. diff --git a/sei-cosmos/server/grpc/server_rate_limit_test.go b/sei-cosmos/server/grpc/server_rate_limit_test.go index cfda06990d..575141f999 100644 --- a/sei-cosmos/server/grpc/server_rate_limit_test.go +++ b/sei-cosmos/server/grpc/server_rate_limit_test.go @@ -126,6 +126,7 @@ func TestStartGRPCServer_RateLimitingEnabledInstallsInterceptors(t *testing.T) { const ( rateLimitRejectedMetric = "rpc_rate_limit_rejected_total" inFlightRejectedMetric = "rpc_inflight_rejected_total" + connRejectedMetric = "rpc_connection_rejected_total" ) var (