router: budget the per-connection protocols from the connection cap

A member attached elsewhere holds a connection to this router for its
DHT queries, with a DHT stream and gossip streams on it like a member's
own, and at 10,000 members the testnet's routers carried 9,000
connections each for 3,600 members. The identify, auth, DHT and gossip
budgets were sized from the member watermark (6,000), so the gossip
streams were refused past it in the last 10k run (327k on one router)
and the DHT streams would have been next. They follow the connection
cap now, as the connections they ride on do.

Author: aojea

internal/router/resource_limits.go

@@ -94,8 +94,13 @@ func hubLimits(highWaterMark, conns, circuits int) rcmgr.ConcreteLimitConfig {
 		circuit.ProtoIDv2Hop:  relayStreams,
 		circuit.ProtoIDv2Stop: relayStreams,
 	}
+	// Every connection, not every member: a member attached elsewhere holds
+	// a connection here for its DHT queries, with a DHT stream and gossip
+	// streams on it, and at 10,000 members the testnet's routers carried
+	// 9,000 connections each for 3,600 members. The connection cap is what
+	// bounds those streams.
 	for _, p := range memberProtocols {
-		protocols[p] = streams(sysBase)
+		protocols[p] = streams(connCap)
 	}
 
 	overrides := rcmgr.PartialLimitConfig{
@@ -105,7 +110,7 @@ func hubLimits(highWaterMark, conns, circuits int) rcmgr.ConcreteLimitConfig {
 		// stream per relayed caller, up to the per-peer circuit budget.
 		PeerDefault: perPeerCircuits,
 		Service: map[string]rcmgr.ResourceLimits{
-			identify.ServiceName: streams(sysBase),
+			identify.ServiceName: streams(connCap),
 			relay.ServiceName:    relayStreams,
 		},
 		ServicePeer: map[string]rcmgr.ResourceLimits{

internal/router/resource_limits_test.go

@@ -49,15 +49,17 @@ func TestHubLimitsFollowTheConnectionBudget(t *testing.T) {
 	want("system inbound streams", cfg.System.StreamsInbound, 8*connCapHeadroom*highWaterMark)
 	want("transient inbound conns", cfg.Transient.ConnsInbound, highWaterMark)
 	want("transient inbound streams", cfg.Transient.StreamsInbound, highWaterMark)
-	want("identify service inbound streams", cfg.Service[identify.ServiceName].StreamsInbound, highWaterMark)
-	want("identify protocol inbound streams", cfg.Protocol[identify.ID].StreamsInbound, highWaterMark)
-	want("auth protocol inbound streams", cfg.Protocol[api.AuthProtocolID].StreamsInbound, highWaterMark)
-	// Every member holds one DHT stream and one gossip stream to its router
-	// for as long as it is connected; libp2p's default capped them at a few
-	// hundred in all and the rest of the fleet fell out of discovery.
-	want("dht protocol inbound streams", cfg.Protocol[api.DHTProtocolID].StreamsInbound, highWaterMark)
+	want("identify service inbound streams", cfg.Service[identify.ServiceName].StreamsInbound, connCapHeadroom*highWaterMark)
+	want("identify protocol inbound streams", cfg.Protocol[identify.ID].StreamsInbound, connCapHeadroom*highWaterMark)
+	want("auth protocol inbound streams", cfg.Protocol[api.AuthProtocolID].StreamsInbound, connCapHeadroom*highWaterMark)
+	// Every connection holds one DHT stream and one gossip stream for as
+	// long as it lasts, and a router carries connections from members
+	// attached elsewhere as well as its own, so these follow the connection
+	// cap; libp2p's default capped them at a few hundred in all and the rest
+	// of the fleet fell out of discovery.
+	want("dht protocol inbound streams", cfg.Protocol[api.DHTProtocolID].StreamsInbound, connCapHeadroom*highWaterMark)
 	for _, p := range pubsub.GossipSubDefaultProtocols {
-		want("gossip protocol "+string(p)+" inbound streams", cfg.Protocol[p].StreamsInbound, highWaterMark)
+		want("gossip protocol "+string(p)+" inbound streams", cfg.Protocol[p].StreamsInbound, connCapHeadroom*highWaterMark)
 	}
 	want("relay service inbound streams", cfg.Service[relay.ServiceName].StreamsInbound, highWaterMark)
 	// yamux reserves one window per open stream through the manager, so