p2plimits: a stream budget carries the memory its streams reserve

yamux reserves one 256 KiB receive window per stream it opens, through
the resource manager, and releases it with the stream; the bytes are
allocated only as data arrives. With the node dialling TCP first, every
member's hub link runs over yamux, and a router's memory budget, which
libp2p scales from host memory and nothing here had set, filled at 1500
members: 1.17 GB reserved of 1.18 GB on the testnet's routers, after
which the relay refused every circuit with RESOURCE_LIMIT_EXCEEDED (3200
to 3400 per router in one ladder step) while every stream counter stayed
at zero, since a refused memory reservation is not a refused stream.

Streams(n) now carries the memory its 2n streams reserve, one window
each, so the system, transient, service and protocol budgets built from
it follow the connection budget as the streams do; the router's and the
node's peer scope get the same budget as the per-peer circuits, since the
peer scope for a service many members call carries one stop stream per
caller. The reporter counts refused memory reservations in
agentmesh_{router,node}_memory_reservations_refused_total and logs them
once a minute, naming the libp2p gauges that show which scope is full.
Tests pin the budgets and check that a manager built from Streams(n)
admits 2n windows and refuses the next.

Author: aojea

internal/node/resource_limits.go

@@ -93,6 +93,9 @@ func nodeLimits(conns int) rcmgr.ConcreteLimitConfig {
 	overrides := rcmgr.PartialLimitConfig{
 		System:    system,
 		Transient: transient,
+		// The peer scope for a router carries one relay stop stream per
+		// caller it relays, as many as the connection budget admits.
+		PeerDefault: p2plimits.Streams(conns),
 		Service: map[string]rcmgr.ResourceLimits{
 			identify.ServiceName: p2plimits.Streams(conns),
 		},
@@ -131,6 +134,13 @@ var streamsRefusedTotal = promauto.NewCounterVec(
 	[]string{"scope", "protocol"},
 )
 
+// memoryRefusedTotal counts the memory reservations the resource manager
+// refused: the muxer's window for a new stream, among others.
+var memoryRefusedTotal = promauto.NewCounter(prometheus.CounterOpts{
+	Name: "agentmesh_node_memory_reservations_refused_total",
+	Help: "Memory reservations the resource manager refused; a new stream that needed one was refused with it",
+})
+
 // busiestConnStreams is the most streams any one connection carries. On a
 // provider the connection to a router carries one stream per caller
 // relayed through it; a QUIC connection admits
@@ -152,5 +162,5 @@ const streamPressureInterval = 15 * time.Second
 // charged to the router's address; see p2plimits.Observed.
 func newResourceManager() (network.ResourceManager, error) {
 	return p2plimits.New(nodeLimits(ConnHighWaterMark), ConnsPerSourceIP, inboundRefusedTotal, logger.Warnf,
-		rcmgr.WithMetrics(p2plimits.CountStreamRefusals(streamsRefusedTotal, logger.Warnf)))
+		rcmgr.WithMetrics(p2plimits.CountStreamRefusals(streamsRefusedTotal, memoryRefusedTotal, logger.Warnf)))
 }

internal/node/resource_limits_test.go

@@ -18,6 +18,7 @@ import (
 	"testing"
 
 	"github.com/google/agentmesh/api"
+	"github.com/google/agentmesh/internal/p2plimits"
 	pubsub "github.com/libp2p/go-libp2p-pubsub"
 	"github.com/libp2p/go-libp2p/core/connmgr"
 	rcmgr "github.com/libp2p/go-libp2p/p2p/host/resource-manager"
@@ -50,6 +51,15 @@ func TestNodeLimitsFollowTheConnectionBudget(t *testing.T) {
 	// Every caller through one router is a stop stream from that one peer.
 	want("relay stop protocol inbound streams", cfg.Protocol[circuit.ProtoIDv2Stop].StreamsInbound, ConnHighWaterMark)
 	want("relay stop protocol per-peer inbound streams", cfg.ProtocolPeer[circuit.ProtoIDv2Stop].StreamsInbound, ConnHighWaterMark)
+	// A router peer carries one stop stream per relayed caller; yamux
+	// reserves one window per open stream, so memory follows the streams.
+	want("peer inbound streams", cfg.PeerDefault.StreamsInbound, ConnHighWaterMark)
+	if got, min := int64(cfg.System.Memory), int64(cfg.System.Streams)*p2plimits.StreamWindow; got < min {
+		t.Errorf("system memory = %d, want at least a window per stream, %d", got, min)
+	}
+	if got, min := int64(cfg.PeerDefault.Memory), int64(ConnHighWaterMark)*p2plimits.StreamWindow; got < min {
+		t.Errorf("peer memory = %d, want at least a window per stop stream, %d", got, min)
+	}
 
 	if ConnsPerSourceIP != ConnHighWaterMark/4 {
 		t.Errorf("ConnsPerSourceIP = %d, want a quarter of the high watermark, %d", ConnsPerSourceIP, ConnHighWaterMark/4)

internal/p2plimits/limits.go

@@ -122,15 +122,27 @@ func sourceRateLimiter(perIP int) *xrate.Limiter {
 	}
 }
 
-// Streams is a stream budget of n each way.
+// Streams is a stream budget of n each way, with the memory the streams
+// reserve: yamux reserves one receive window per open stream through the
+// resource manager, so a budget of streams is a budget of memory, and
+// libp2p's default for the latter is scaled from host memory, which on the
+// testnet's routers was 1.1 GiB and full at 1500 members over TCP. The
+// memory is a reservation, released with the stream, and allocated only
+// as data arrives and waits to be read.
 func Streams(n int) rcmgr.ResourceLimits {
 	return rcmgr.ResourceLimits{
 		Streams:         rcmgr.LimitVal(2 * n),
 		StreamsInbound:  rcmgr.LimitVal(n),
 		StreamsOutbound: rcmgr.LimitVal(n),
+		Memory:          rcmgr.LimitVal64(int64(2*n) * StreamWindow),
 	}
 }
 
+// StreamWindow is the receive window yamux reserves for a stream when it
+// opens (go-yamux's initialStreamWindow); a stream that carries a fast
+// transfer to a slow reader may grow it, up to the muxer's maximum.
+const StreamWindow = 256 << 10
+
 // RefusalLogInterval bounds how often one source IP's refusals are logged;
 // the counter carries the rate.
 const RefusalLogInterval = time.Minute

internal/p2plimits/streams.go

@@ -15,6 +15,7 @@
 package p2plimits
 
 import (
+	"fmt"
 	"sync"
 	"time"
 
@@ -40,32 +41,37 @@ const (
 )
 
 // StreamRefusals counts the streams the resource manager refused, by scope
-// and protocol (or service), and logs each scope and name once per
-// RefusalLogInterval. It is libp2p's MetricsReporter; the Allow side is a
-// no-op.
+// and protocol (or service), and the memory reservations it refused, and
+// logs each once per RefusalLogInterval. It is libp2p's MetricsReporter;
+// the Allow side is a no-op.
 type StreamRefusals struct {
-	refused   *prometheus.CounterVec
-	warnf     func(format string, args ...any)
-	mu        sync.Mutex
-	since     map[string]uint64
-	lastLogAt map[string]time.Time
-	now       func() time.Time
+	refused       *prometheus.CounterVec
+	memoryRefused prometheus.Counter
+	warnf         func(format string, args ...any)
+	mu            sync.Mutex
+	since         map[string]uint64
+	lastLogAt     map[string]time.Time
+	now           func() time.Time
 }
 
 var _ rcmgr.MetricsReporter = (*StreamRefusals)(nil)
 
 // CountStreamRefusals returns a reporter for rcmgr.WithMetrics. refused
-// carries the labels scope and protocol; a nil warnf only counts.
-func CountStreamRefusals(refused *prometheus.CounterVec, warnf func(format string, args ...any)) *StreamRefusals {
+// carries the labels scope and protocol; memoryRefused counts the memory
+// reservations refused, which the stream muxer makes one of per stream it
+// opens and the relay one of per connection it relays; a nil warnf only
+// counts.
+func CountStreamRefusals(refused *prometheus.CounterVec, memoryRefused prometheus.Counter, warnf func(format string, args ...any)) *StreamRefusals {
 	if warnf == nil {
 		warnf = func(string, ...any) {}
 	}
 	return &StreamRefusals{
-		refused:   refused,
-		warnf:     warnf,
-		since:     map[string]uint64{},
-		lastLogAt: map[string]time.Time{},
-		now:       time.Now,
+		refused:       refused,
+		memoryRefused: memoryRefused,
+		warnf:         warnf,
+		since:         map[string]uint64{},
+		lastLogAt:     map[string]time.Time{},
+		now:           time.Now,
 	}
 }
 
@@ -74,7 +80,31 @@ func (s *StreamRefusals) SetClock(now func() time.Time) { s.now = now }
 
 func (s *StreamRefusals) refusedStream(scope, name string) {
 	s.refused.WithLabelValues(scope, name).Inc()
-	key := scope + " " + name
+	s.report(scope+" "+name, func(n uint64) string {
+		what := "at the " + scope + " scope"
+		if name != "" {
+			what = "on " + name + " " + what
+		}
+		return fmt.Sprintf("[Stream] Refused %d stream(s) %s since the last report; the peer sees its stream reset", n, what)
+	})
+}
+
+// BlockMemory is a memory reservation the manager refused: the muxer's
+// window for a new stream, or the relay's buffers for a connection it
+// would relay. The budget it ran into is libp2p_rcmgr_limit{resource=
+// "memory"} on the scope libp2p_rcmgr_memory shows full.
+func (s *StreamRefusals) BlockMemory(size int) {
+	if s.memoryRefused != nil {
+		s.memoryRefused.Inc()
+	}
+	s.report("memory", func(n uint64) string {
+		return fmt.Sprintf("[Memory] Refused %d memory reservation(s) since the last report, the last for %d bytes: the resource manager's memory budget is full (libp2p_rcmgr_memory against libp2p_rcmgr_limit{resource=\"memory\"}); a new stream or a relayed connection that needs it is refused", n, size)
+	})
+}
+
+// report counts one refusal under key and, once per RefusalLogInterval,
+// logs the line for the count since the last report.
+func (s *StreamRefusals) report(key string, line func(n uint64) string) {
 	s.mu.Lock()
 	s.since[key]++
 	n := s.since[key]
@@ -87,11 +117,7 @@ func (s *StreamRefusals) refusedStream(scope, name string) {
 	}
 	s.mu.Unlock()
 	if shouldLog {
-		what := "at the " + scope + " scope"
-		if name != "" {
-			what = "on " + name + " " + what
-		}
-		s.warnf("[Stream] Refused %d stream(s) %s since the last report; the peer sees its stream reset", n, what)
+		s.warnf("%s", line(n))
 	}
 }
 
@@ -124,4 +150,3 @@ func (s *StreamRefusals) AllowPeer(peer.ID)                      {}
 func (s *StreamRefusals) AllowProtocol(protocol.ID)              {}
 func (s *StreamRefusals) AllowService(string)                    {}
 func (s *StreamRefusals) AllowMemory(int)                        {}
-func (s *StreamRefusals) BlockMemory(int)                        {}

internal/p2plimits/streams_test.go

@@ -16,13 +16,15 @@ package p2plimits
 
 import (
 	"fmt"
+	"strings"
 	"testing"
 	"time"
 
 	"github.com/libp2p/go-libp2p/core/network"
 	"github.com/libp2p/go-libp2p/core/protocol"
 	"github.com/libp2p/go-libp2p/core/test"
 	rcmgr "github.com/libp2p/go-libp2p/p2p/host/resource-manager"
+	"github.com/multiformats/go-multiaddr"
 	"github.com/prometheus/client_golang/prometheus"
 )
 
@@ -38,7 +40,7 @@ func TestStreamRefusalsCountTheProtocol(t *testing.T) {
 		Protocol: map[protocol.ID]rcmgr.ResourceLimits{proto: Streams(1)},
 	}.Build(rcmgr.InfiniteLimits)
 	refused := newStreamsRefusedCounter()
-	m, err := NewResourceManager(limits, rcmgr.WithMetrics(CountStreamRefusals(refused, nil)))
+	m, err := NewResourceManager(limits, rcmgr.WithMetrics(CountStreamRefusals(refused, nil, nil)))
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -70,7 +72,7 @@ func TestStreamRefusalsCountTheProtocol(t *testing.T) {
 func TestStreamRefusalsLogOneProtocolOncePerInterval(t *testing.T) {
 	now := time.Unix(1_700_000_000, 0)
 	var lines []string
-	s := CountStreamRefusals(newStreamsRefusedCounter(), func(format string, args ...any) {
+	s := CountStreamRefusals(newStreamsRefusedCounter(), nil, func(format string, args ...any) {
 		lines = append(lines, fmt.Sprintf(format, args...))
 	})
 	s.SetClock(func() time.Time { return now })
@@ -96,3 +98,61 @@ func TestStreamRefusalsLogOneProtocolOncePerInterval(t *testing.T) {
 		t.Errorf("peer scope: %q", lines[len(lines)-1])
 	}
 }
+
+// A refused memory reservation is counted and said once per interval; the
+// muxer makes one per stream it opens, so a full memory budget is a stream
+// the peer never gets, with nothing in the stream counters.
+func TestStreamRefusalsCountRefusedMemory(t *testing.T) {
+	memory := prometheus.NewCounter(prometheus.CounterOpts{Name: "test_memory_refused_total"})
+	var lines []string
+	now := time.Unix(1_700_000_000, 0)
+	s := CountStreamRefusals(newStreamsRefusedCounter(), memory, func(format string, args ...any) {
+		lines = append(lines, fmt.Sprintf(format, args...))
+	})
+	s.SetClock(func() time.Time { return now })
+	for i := 0; i < 3; i++ {
+		s.BlockMemory(256 << 10)
+	}
+	if got := counterValue(t, memory); got != 3 {
+		t.Errorf("memory refusals counted = %v, want 3", got)
+	}
+	if len(lines) != 1 || !strings.Contains(lines[0], "Refused 1 memory reservation(s)") || !strings.Contains(lines[0], "262144 bytes") {
+		t.Fatalf("lines = %q", lines)
+	}
+	now = now.Add(RefusalLogInterval)
+	s.BlockMemory(4096)
+	if len(lines) != 2 || !strings.Contains(lines[1], "Refused 3 memory reservation(s)") {
+		t.Errorf("after the interval: %q", lines)
+	}
+}
+
+// The memory a stream budget carries admits the streams it admits: a
+// manager built from Streams(n) opens n streams on one yamux-sized
+// reservation each, and refuses the reservation past the budget.
+func TestStreamsBudgetCarriesTheMuxerWindows(t *testing.T) {
+	const n = 4
+	limits := rcmgr.PartialLimitConfig{System: Streams(n), Transient: Streams(n), PeerDefault: Streams(n)}.Build(rcmgr.InfiniteLimits)
+	m, err := NewResourceManager(limits)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer func() { _ = m.Close() }()
+	p := test.RandPeerIDFatal(t)
+	conn, err := m.OpenConnection(network.DirInbound, true, multiaddr.StringCast("/ip4/203.0.113.9/tcp/1"))
+	if err != nil {
+		t.Fatal(err)
+	}
+	if err := conn.SetPeer(p); err != nil {
+		t.Fatal(err)
+	}
+	// Streams(n) admits 2n streams in all; each takes one window, as yamux
+	// reserves it, so the 2n+1-th window is refused.
+	for i := 0; i < 2*n; i++ {
+		if err := conn.ReserveMemory(StreamWindow, network.ReservationPriorityAlways); err != nil {
+			t.Fatalf("window %d refused: %v", i+1, err)
+		}
+	}
+	if err := conn.ReserveMemory(StreamWindow, network.ReservationPriorityAlways); err == nil {
+		t.Fatal("a window past the stream budget was admitted")
+	}
+}

internal/router/resource_limits.go

@@ -85,9 +85,9 @@ func hubLimits(highWaterMark, conns, circuits int) rcmgr.ConcreteLimitConfig {
 
 	// A relayed connection is one hop stream from the caller and one stop
 	// stream to the destination, both held while it lasts; the memory the
-	// relay reserves per connection is a few KiB on each.
+	// relay reserves per connection is a few KiB on each, on top of the
+	// streams' windows.
 	relayStreams := streams(sysBase)
-	relayStreams.Memory = rcmgr.LimitVal64(int64(sysBase) * 64 << 10)
 	perPeerCircuits := streams(circuits)
 
 	protocols := map[protocol.ID]rcmgr.ResourceLimits{
@@ -101,6 +101,9 @@ func hubLimits(highWaterMark, conns, circuits int) rcmgr.ConcreteLimitConfig {
 	overrides := rcmgr.PartialLimitConfig{
 		System:    system,
 		Transient: transient,
+		// The peer scope for a service many members call carries one stop
+		// stream per relayed caller, up to the per-peer circuit budget.
+		PeerDefault: perPeerCircuits,
 		Service: map[string]rcmgr.ResourceLimits{
 			identify.ServiceName: streams(sysBase),
 			relay.ServiceName:    relayStreams,

internal/router/resource_limits_test.go

@@ -19,6 +19,7 @@ import (
 	"testing"
 
 	"github.com/google/agentmesh/api"
+	"github.com/google/agentmesh/internal/p2plimits"
 	dht "github.com/libp2p/go-libp2p-kad-dht"
 	pubsub "github.com/libp2p/go-libp2p-pubsub"
 	rcmgr "github.com/libp2p/go-libp2p/p2p/host/resource-manager"
@@ -59,6 +60,18 @@ func TestHubLimitsFollowTheConnectionBudget(t *testing.T) {
 		want("gossip protocol "+string(p)+" inbound streams", cfg.Protocol[p].StreamsInbound, highWaterMark)
 	}
 	want("relay service inbound streams", cfg.Service[relay.ServiceName].StreamsInbound, highWaterMark)
+	// yamux reserves one window per open stream through the manager, so
+	// the memory budget follows the stream budget; libp2p's default is
+	// scaled from host memory and was full at 1500 members over TCP.
+	if got, min := int64(cfg.System.Memory), int64(cfg.System.Streams)*p2plimits.StreamWindow; got < min {
+		t.Errorf("system memory = %d, want at least a window per stream, %d", got, min)
+	}
+	if got, min := int64(cfg.PeerDefault.Memory), int64(circuits)*p2plimits.StreamWindow; got < min {
+		t.Errorf("peer memory = %d, want at least a window per circuit, %d", got, min)
+	}
+	// A service many members call holds one stop stream per caller on the
+	// router's peer scope for it.
+	want("peer outbound streams", cfg.PeerDefault.StreamsOutbound, circuits)
 	want("hop protocol inbound streams", cfg.Protocol[circuit.ProtoIDv2Hop].StreamsInbound, highWaterMark)
 	// A destination many members call holds one stop stream per caller.
 	want("relay per-peer outbound streams", cfg.ServicePeer[relay.ServiceName].StreamsOutbound, circuits)

internal/router/resource_observer.go

@@ -40,6 +40,14 @@ var streamsRefusedTotal = promauto.NewCounterVec(
 	[]string{"scope", "protocol"},
 )
 
+// memoryRefusedTotal counts the memory reservations the resource manager
+// refused: the muxer's window for a new stream, the relay's buffers for a
+// connection. The budget follows the stream budget (see p2plimits.Streams).
+var memoryRefusedTotal = promauto.NewCounter(prometheus.CounterOpts{
+	Name: "agentmesh_router_memory_reservations_refused_total",
+	Help: "Memory reservations the resource manager refused; a new stream or a relayed connection that needed one was refused with it",
+})
+
 // busiestConnStreams is the most streams any one connection carries. A
 // QUIC connection admits p2plimits.QUICStreamsPerConnection; the router's
 // connection to a service it relays for carries one stream per relayed
@@ -67,7 +75,7 @@ func init() {
 // countStreamRefusals is the reporter the resource manager is built with;
 // it counts refused streams in streamsRefusedTotal by protocol.
 func countStreamRefusals() *p2plimits.StreamRefusals {
-	return p2plimits.CountStreamRefusals(streamsRefusedTotal, logger.Warnf)
+	return p2plimits.CountStreamRefusals(streamsRefusedTotal, memoryRefusedTotal, logger.Warnf)
 }
 
 // connsPerSourceIPDesc exports the cap in force, so a dashboard can read the

site/content/docs/guides/scalability.md

@@ -101,6 +101,22 @@ from host memory. A refused stream is counted in
 `agentmesh_router_streams_refused_total{protocol}` and
 `agentmesh_node_streams_refused_total{protocol}`.
 
+A stream over TCP also reserves memory. yamux reserves one receive window
+(256 KiB) per stream it opens, through the resource manager, and releases
+it when the stream closes; the memory is allocated only as data arrives
+and waits to be read, so an idle stream holds a reservation and no bytes.
+The memory budget therefore follows the stream budget, one window per
+stream the budget admits, on the system scope and on the scope of each
+peer. libp2p's own default for the system scope is scaled from host
+memory and was 1.1 GiB on the testnet's routers, full at 1500 members
+over TCP, at which point the relay refused every circuit with
+`RESOURCE_LIMIT_EXCEEDED` and no stream counter moved. A refused
+reservation is counted in
+`agentmesh_router_memory_reservations_refused_total` and
+`agentmesh_node_memory_reservations_refused_total`, and the scope that is
+full reads from `libp2p_rcmgr_memory{scope}` against
+`libp2p_rcmgr_limit{resource="memory",scope}`.
+
 ### Hub links and peer links
 
 A member's connection to a router is a hub link. It carries the member's
@@ -367,6 +383,7 @@ The router and the control plane serve Prometheus metrics on
 | `libp2p_relaysvc_reservations_total{type="opened"}` minus `{type="closed"}`, against `agentmesh_router_relay_limit{limit="reservations"}` | Reservations in use against the budget. |
 | `libp2p_relaysvc_connection_rejections_total{reason="resource limit exceeded"}` | A service has more callers than `--relay-max-circuits`. |
 | `libp2p_rcmgr_blocked_resources` | libp2p refused a connection or stream on a scope. On a router this should stay flat. |
+| `agentmesh_router_memory_reservations_refused_total`, `agentmesh_node_memory_reservations_refused_total`, with `libp2p_rcmgr_memory{scope}` against `libp2p_rcmgr_limit{resource="memory",scope}` | The resource manager's memory budget on some scope is full. Over TCP every open stream holds a 256 KiB reservation, so this is the stream budget met from the other side: new streams and relayed connections are refused, with `RESOURCE_LIMIT_EXCEEDED` at the relay and nothing in the stream counters. The scope whose `libp2p_rcmgr_memory` is at its limit names the budget. |
 | `agentmesh_router_streams_refused_total{scope,protocol}`, `agentmesh_node_streams_refused_total{scope,protocol}` | The streams behind that gauge, by protocol. `/mesh/kad/1.0.0` or `/meshsub/*` at the `protocol` scope on a router is members past the stream budget of a protocol, which keeps them out of discovery; `agentmesh_router_dht_routing_table_size` falls with it. |
 | `agentmesh_router_auth_handshakes_total{result="read_failed"}` | Members that timed out mid-handshake: the router is slow, usually CPU. |
 | `agentmesh_router_draining` | The router is stopping. Exclude it from capacity sums. |