chore(rpc): close slow WebSocket clients by default (defense-in-depth) (#147)

* chore(rpc): close slow WebSocket clients by default

Sets CometBFT's experimental_close_on_slow_client=true via initCometBFTConfig
so any WebSocket subscriber that cannot keep up with its subscription buffer
is forcibly disconnected by the server. This is defense-in-depth against the
failure mode observed on lumera-devnet-1 val3, where a client-side socket
leak in sdk-go (fixed in LumeraProtocol/sdk-go#17) accumulated ~5,000
ESTABLISHED WS connections, saturated the RPC listen backlog
(Recv-Q=4097/4096), and made external port :26687 unresponsive — while the
chain itself kept producing blocks.

The sdk-go fix stops our own client from leaking. This change protects
mainnet validators from any third-party client (indexer, relayer, custom
tooling) exhibiting the same pattern.

Subscription caps (max_subscription_clients=100, max_subscriptions_per_client=5)
are left at CometBFT defaults; the new test locks them in as a guard against
accidental relaxation that would re-open the saturation surface.

Trade-off: well-behaved clients on high-volume subscriptions that briefly
stall longer than experimental_subscription_buffer_size events will be
disconnected and must reconnect. Reconnect handling is already required for
operational reasons (validator restart, network blips); this is not a new
client-side requirement.

* test(system): suppress expected node stop exits

Author: mateeullahmalik

cmd/lumera/cmd/config.go

@@ -72,6 +72,38 @@ func initCometBFTConfig() *cmtcfg.Config {
 	// cfg.P2P.MaxNumInboundPeers = 100
 	// cfg.P2P.MaxNumOutboundPeers = 40
 
+	// RPC hardening — defense-in-depth against misbehaving WebSocket clients.
+	//
+	// Default CometBFT behaviour keeps a WebSocket connection open even when the
+	// client is not draining its subscription buffer, relying on the OS / peer to
+	// eventually close the socket. In practice this can take hours, and a buggy
+	// client (or one with a goroutine/socket leak — see sdk-go PR #17) can
+	// accumulate thousands of dormant subscriptions on `:26657` and saturate the
+	// listen backlog of a validator's RPC.
+	//
+	// Observed on lumera-devnet-1 val3: ~5,000 ESTABLISHED WS connections, listen
+	// queue full (Recv-Q=4097/4096), external RPC port unresponsive — while the
+	// chain itself kept producing blocks. Root cause was a client-side leak in
+	// sdk-go's wait-tx subscriber (fixed in sdk-go PR #17), but the server had no
+	// circuit breaker: it accepted, kept, and buffered for every slow subscriber
+	// indefinitely.
+	//
+	// Setting CloseOnSlowClient=true tells CometBFT to forcibly close any WS
+	// subscriber that cannot keep up with its subscription buffer. This is the
+	// upstream-recommended defense against the failure mode and protects mainnet
+	// validators from any third-party client (indexer, relayer, custom tooling)
+	// that exhibits the same pattern — sdk-go-based or otherwise.
+	//
+	// Trade-off: well-behaved clients that subscribe to a high-volume query and
+	// briefly stall (network blip, GC pause >subscription_buffer_size events)
+	// will be disconnected and must reconnect. They already had to handle
+	// reconnect for normal operational reasons (validator restart, network), so
+	// this is not a new client-side requirement.
+	//
+	// Subscription caps are left at CometBFT defaults (100 clients × 5 subs each),
+	// which are appropriate; no live evidence justifies changing them.
+	cfg.RPC.CloseOnSlowClient = true
+
 	return cfg
 }
 

cmd/lumera/cmd/config_test.go

@@ -53,3 +53,30 @@ func TestInitAppConfigEVMDefaults(t *testing.T) {
 	require.True(t, evmMempoolCfg.IsValid(), "Lumera.EVMMempool field not found")
 	require.False(t, evmMempoolCfg.FieldByName("BroadcastDebug").Bool(), "broadcast debug must be disabled by default")
 }
+
+// TestInitCometBFTConfigRPCHardening locks in the RPC defense-in-depth
+// overrides applied in initCometBFTConfig. These protect lumerad's RPC
+// (port :26657) from misbehaving WebSocket clients accumulating dormant
+// subscriptions (see sdk-go PR #17 / lumera-devnet-1 val3 incident, where
+// ~5,000 ESTABLISHED conns saturated the listen backlog while the chain
+// itself stayed healthy). Changing these defaults should require an
+// explicit decision recorded against this test.
+func TestInitCometBFTConfigRPCHardening(t *testing.T) {
+	t.Parallel()
+
+	cfg := initCometBFTConfig()
+
+	require.NotNil(t, cfg, "initCometBFTConfig must return a config")
+	require.NotNil(t, cfg.RPC, "RPC config must not be nil")
+
+	require.True(t, cfg.RPC.CloseOnSlowClient,
+		"experimental_close_on_slow_client must default to true to prevent slow WS clients from holding subscription buffers indefinitely")
+
+	// Subscription caps are upstream CometBFT defaults — we intentionally do
+	// not narrow them; this assertion guards against accidentally loosening
+	// them, which would re-open the saturation attack surface.
+	require.Equal(t, 100, cfg.RPC.MaxSubscriptionClients,
+		"max_subscription_clients must stay at the upstream default; relaxing it re-opens the val3-style saturation surface")
+	require.Equal(t, 5, cfg.RPC.MaxSubscriptionsPerClient,
+		"max_subscriptions_per_client must stay at the upstream default")
+}

tests/systemtests/system.go

@@ -68,8 +68,8 @@ type SystemUnderTest struct {
 	verbose           bool
 	ChainStarted      bool
 	projectName       string
-	dirty             bool // requires full reset when marked dirty
-	stopping          bool // true while StopChain is running (suppresses exit errors)
+	dirty             bool        // requires full reset when marked dirty
+	stopping          atomic.Bool // true while StopChain is running (suppresses expected exit errors)
 
 	pidsLock mtxSync.RWMutex
 	pids     map[int]struct{}
@@ -328,7 +328,7 @@ func (s *SystemUnderTest) StopChain() {
 	if !s.ChainStarted {
 		return
 	}
-	s.stopping = true
+	s.stopping.Store(true)
 
 	// Pre-cleanup: unsubscribe from events while nodes are still alive
 	for _, c := range s.cleanupPreFn {
@@ -374,7 +374,7 @@ func (s *SystemUnderTest) StopChain() {
 	s.cleanupPostFn = nil
 
 	s.ChainStarted = false
-	s.stopping = false
+	s.stopping.Store(false)
 }
 
 func (s *SystemUnderTest) withEachPid(cb func(p *os.Process)) {
@@ -636,7 +636,7 @@ func (s *SystemUnderTest) startNodesAsync(t *testing.T, xargs ...string) {
 		go func(pid int, cmd *exec.Cmd, nodeIndex int) {
 			defer wg.Done()
 			err := cmd.Wait() // blocks until shutdown
-			if err != nil {
+			if err != nil && !s.stopping.Load() {
 				s.PrintBuffer()
 				errChan <- fmt.Errorf("node %d exited unexpectedly: %w", nodeIndex, err)
 			} else {
@@ -653,13 +653,13 @@ func (s *SystemUnderTest) startNodesAsync(t *testing.T, xargs ...string) {
 	// Wait in background and close the channel when all nodes are done
 	go func() {
 		wg.Wait()
+		close(errChan)
 
 		for err := range errChan {
-			if err != nil && !s.stopping {
+			if err != nil {
 				t.Errorf("%v", err)
 			}
 		}
-		close(errChan)
 	}()
 }