From 00b46317f0e7e842eeec262ed2c0024f8b85c878 Mon Sep 17 00:00:00 2001 From: kalo <24719519+KaloyanTanev@users.noreply.github.com> Date: Thu, 3 Sep 2026 16:18:31 +0300 Subject: [PATCH 1/7] p2p: add message size and concurrency metrics --- docs/metrics.md | 6 ++ p2p/metrics.go | 98 +++++++++++++++++++++++ p2p/metrics_internal_test.go | 149 +++++++++++++++++++++++++++++++++++ p2p/receive.go | 11 +++ p2p/sender.go | 7 ++ 5 files changed, 271 insertions(+) create mode 100644 p2p/metrics_internal_test.go diff --git a/docs/metrics.md b/docs/metrics.md index 97d28d849d..9b4e242993 100644 --- a/docs/metrics.md +++ b/docs/metrics.md @@ -99,6 +99,10 @@ when storing metrics from multiple nodes or clusters in one Prometheus instance. | `core_validatorapi_request_latency_seconds` | Histogram | The validatorapi request latencies in seconds by endpoint | `endpoint` | | `core_validatorapi_request_total` | Counter | The total number of requests per content-type and endpoint | `endpoint, content_type` | | `core_validatorapi_vc_user_agent` | Gauge | Gauge with label set to user agent string of requests made by VC | `user_agent` | +| `p2p_concurrent_requests` | Histogram | Number of concurrently handled inbound messages, observed at each message arrival, by protocol and sending peer. Unlike the sampled inflight_requests gauge, this captures bursts between scrapes. | `protocol, peer` | +| `p2p_handler_duration_seconds` | Histogram | Duration of inbound libp2p message handling from stream accept to handler completion by protocol. Explains inflight_requests: inflight equals message rate times this duration. | `protocol` | +| `p2p_inflight_requests` | Gauge | Current number of inbound libp2p messages being handled (stream accept to handler completion) by protocol and sending peer. | `protocol, peer` | +| `p2p_message_read_errors_total` | Counter | Total number of failures reading a libp2p message by protocol and sending peer. Includes messages exceeding the protocol read limit. | `protocol, peer` | | `p2p_peer_connection_total` | Counter | Total number of libp2p connections per peer. | `peer` | | `p2p_peer_connection_types` | Gauge | Current number of libp2p connections by peer, type (`direct` or `relay`), and protocol (`tcp`, `quic`). Note that peers may have multiple connections. | `peer, type, protocol` | | `p2p_peer_network_receive_bytes_total` | Counter | Total number of network bytes received from the peer by protocol and transport. Transport is based on first active connection (accurate in steady state). | `peer, protocol, transport` | @@ -108,9 +112,11 @@ when storing metrics from multiple nodes or clusters in one Prometheus instance. | `p2p_ping_latency_secs` | Histogram | Ping latencies in seconds per peer | `peer` | | `p2p_ping_success` | Gauge | Whether the last ping was successful (1) or not (0). Can be used as proxy for connected peers | `peer` | | `p2p_reachability_status` | Gauge | Current libp2p reachability status of this node as detected by autonat: unknown(0), public(1) or private(2). | | +| `p2p_received_message_size_bytes` | Histogram | Size in bytes of received libp2p protobuf messages by protocol and sending peer. | `protocol, peer` | | `p2p_relay_connection_types` | Gauge | Current number of libp2p connections by relay, type (`direct` or `relay`), and protocol (`tcp`, `quic`). Note that peers may have multiple connections. | `peer, type, protocol` | | `p2p_relay_connections` | Gauge | Connected relays by name | `peer` | | `p2p_send_duration_seconds` | Histogram | Wall-clock duration of synchronous libp2p Send (one-way) and SendReceive (round-trip) calls, by peer, protocol, and topic. Topic is a sub-protocol label (e.g. qbft_pre_prepare, parsigex_proposer); empty when not set by the caller. | `peer, protocol, topic` | +| `p2p_sent_message_size_bytes` | Histogram | Size in bytes of sent libp2p protobuf messages by protocol. Not labelled by peer since duty messages are broadcast identically to all peers. | `protocol` | | `relay_p2p_active_connections` | Gauge | Current number of active connections by peer and cluster | `peer, peer_cluster` | | `relay_p2p_connection_total` | Counter | Total number of new connections by peer and cluster | `peer, peer_cluster` | | `relay_p2p_network_receive_bytes_total` | Counter | Total number of network bytes received from the peer and cluster | `peer, peer_cluster` | diff --git a/p2p/metrics.go b/p2p/metrics.go index 6f4a04398c..0694c32457 100644 --- a/p2p/metrics.go +++ b/p2p/metrics.go @@ -3,6 +3,7 @@ package p2p import ( + "sync" "time" "github.com/libp2p/go-libp2p" @@ -12,6 +13,7 @@ import ( "github.com/libp2p/go-libp2p/core/protocol" "github.com/libp2p/go-libp2p/p2p/net/swarm" "github.com/prometheus/client_golang/prometheus" + "google.golang.org/protobuf/proto" "github.com/obolnetwork/charon/app/promauto" ) @@ -99,8 +101,104 @@ var ( Name: "peer_network_sent_bytes_total", Help: "Total number of network bytes sent to the peer by protocol and transport. Transport is based on first active connection (accurate in steady state).", }, []string{"peer", "protocol", "transport"}) + + receivedMsgSizeHist = promauto.NewHistogramVec(prometheus.HistogramOpts{ + Namespace: "p2p", + Name: "received_message_size_bytes", + Help: "Size in bytes of received libp2p protobuf messages by protocol and sending peer.", + Buckets: messageSizeBuckets, + }, []string{"protocol", "peer"}) + + sentMsgSizeHist = promauto.NewHistogramVec(prometheus.HistogramOpts{ + Namespace: "p2p", + Name: "sent_message_size_bytes", + Help: "Size in bytes of sent libp2p protobuf messages by protocol. Not labelled by peer since duty messages are broadcast identically to all peers.", + Buckets: messageSizeBuckets, + }, []string{"protocol"}) + + msgReadErrorCounter = promauto.NewCounterVec(prometheus.CounterOpts{ + Namespace: "p2p", + Name: "message_read_errors_total", + Help: "Total number of failures reading a libp2p message by protocol and sending peer. Includes messages exceeding the protocol read limit.", + }, []string{"protocol", "peer"}) + + inflightGauge = promauto.NewGaugeVec(prometheus.GaugeOpts{ + Namespace: "p2p", + Name: "inflight_requests", + Help: "Current number of inbound libp2p messages being handled (stream accept to handler completion) by protocol and sending peer.", + }, []string{"protocol", "peer"}) + + concurrentRequestsHist = promauto.NewHistogramVec(prometheus.HistogramOpts{ + Namespace: "p2p", + Name: "concurrent_requests", + Help: "Number of concurrently handled inbound messages, observed at each message arrival, by protocol and sending peer. Unlike the sampled inflight_requests gauge, this captures bursts between scrapes.", + Buckets: []float64{1, 2, 4, 8, 16, 32, 64, 128, 256}, + }, []string{"protocol", "peer"}) + + handlerDuration = promauto.NewHistogramVec(prometheus.HistogramOpts{ + Namespace: "p2p", + Name: "handler_duration_seconds", + Help: "Duration of inbound libp2p message handling from stream accept to handler completion by protocol. Explains inflight_requests: inflight equals message rate times this duration.", + Buckets: prometheus.ExponentialBuckets(0.001, 2, 15), // 1ms .. ~16s, covers the ~10s default receive timeout. + }, []string{"protocol"}) ) +// inflightCounts tracks the number of concurrently handled inbound messages per (protocol, peer). +// It is the source of truth for both inflightGauge and concurrentRequestsHist so they never drift. +var inflightCounts = struct { + sync.Mutex + + counts map[[2]string]int +}{counts: make(map[[2]string]int)} + +// observeHandlerStart records the start of inbound message handling and returns a done function +// to be called (deferred) when handling completes. +func observeHandlerStart(pID protocol.ID, peerID peer.ID) func() { + key := [2]string{string(pID), PeerName(peerID)} + t0 := time.Now() + + inflightCounts.Lock() + inflightCounts.counts[key]++ + n := inflightCounts.counts[key] + inflightCounts.Unlock() + + inflightGauge.WithLabelValues(key[0], key[1]).Set(float64(n)) + concurrentRequestsHist.WithLabelValues(key[0], key[1]).Observe(float64(n)) + + return func() { + inflightCounts.Lock() + inflightCounts.counts[key]-- + n := inflightCounts.counts[key] + inflightCounts.Unlock() + + inflightGauge.WithLabelValues(key[0], key[1]).Set(float64(n)) + handlerDuration.WithLabelValues(key[0]).Observe(time.Since(t0).Seconds()) + } +} + +// messageSizeBuckets covers charon p2p message sizes from small control messages up to the +// 128MiB default read limit. The 32MiB and 128MiB edges match protocol read limits, so +// messages that (would) exceed a limit are countable by exact bucket subtraction. +var messageSizeBuckets = []float64{ + 256, 1 << 10, 4 << 10, 16 << 10, 64 << 10, 256 << 10, 512 << 10, + 1 << 20, 2 << 20, 4 << 20, 8 << 20, 16 << 20, 32 << 20, 64 << 20, 128 << 20, +} + +// observeReceivedMessage records the size of a successfully read libp2p protobuf message. +func observeReceivedMessage(pID protocol.ID, peerID peer.ID, msg proto.Message) { + receivedMsgSizeHist.WithLabelValues(string(pID), PeerName(peerID)).Observe(float64(proto.Size(msg))) +} + +// observeSentMessage records the size of a successfully written libp2p protobuf message. +func observeSentMessage(pID protocol.ID, msg proto.Message) { + sentMsgSizeHist.WithLabelValues(string(pID)).Observe(float64(proto.Size(msg))) +} + +// incMessageReadError increments the read error counter for the protocol and sending peer. +func incMessageReadError(pID protocol.ID, peerID peer.ID) { + msgReadErrorCounter.WithLabelValues(string(pID), PeerName(peerID)).Inc() +} + func observePing(p peer.ID, d time.Duration) { pingLatencies.WithLabelValues(PeerName(p)).Observe(d.Seconds()) pingSuccess.WithLabelValues(PeerName(p)).Set(1) diff --git a/p2p/metrics_internal_test.go b/p2p/metrics_internal_test.go new file mode 100644 index 0000000000..e7e25492b7 --- /dev/null +++ b/p2p/metrics_internal_test.go @@ -0,0 +1,149 @@ +// Copyright © 2022-2026 Obol Labs Inc. Licensed under the terms of a Business Source License 1.1 + +package p2p + +import ( + "context" + "testing" + "time" + + "github.com/libp2p/go-libp2p/core/peer" + "github.com/libp2p/go-libp2p/core/peerstore" + "github.com/libp2p/go-libp2p/core/protocol" + "github.com/prometheus/client_golang/prometheus" + promtestutil "github.com/prometheus/client_golang/prometheus/testutil" + pb "github.com/prometheus/client_model/go" + "github.com/stretchr/testify/require" + "google.golang.org/protobuf/proto" + + pbv1 "github.com/obolnetwork/charon/core/corepb/v1" + "github.com/obolnetwork/charon/testutil" +) + +// histSample returns the sample count and sum of the histogram for the given label values. +func histSample(t *testing.T, hist *prometheus.HistogramVec, labels ...string) (uint64, float64) { + t.Helper() + + m := new(pb.Metric) + + h, err := hist.GetMetricWithLabelValues(labels...) + require.NoError(t, err) + require.NoError(t, h.(prometheus.Histogram).Write(m)) + + return m.GetHistogram().GetSampleCount(), m.GetHistogram().GetSampleSum() +} + +func TestMessageSizeMetrics(t *testing.T) { + var ( + pID = protocol.ID("test-msg-size") + ctx = context.Background() + client = testutil.CreateHost(t, testutil.AvailableAddr(t)) + server = testutil.CreateHost(t, testutil.AvailableAddr(t)) + ) + + client.Peerstore().AddAddrs(server.ID(), server.Addrs(), peerstore.PermanentAddrTTL) + + RegisterHandler("server", server, pID, + func() proto.Message { return new(pbv1.Duty) }, + func(_ context.Context, _ peer.ID, req proto.Message) (proto.Message, bool, error) { + return req, true, nil + }, + ) + + req := &pbv1.Duty{Slot: 123, Type: 4} + msgSize := float64(proto.Size(req)) + require.Positive(t, msgSize) + + sentCount0, sentSum0 := histSample(t, sentMsgSizeHist, string(pID)) + recvClient0, _ := histSample(t, receivedMsgSizeHist, string(pID), PeerName(server.ID())) + recvServer0, _ := histSample(t, receivedMsgSizeHist, string(pID), PeerName(client.ID())) + + resp := new(pbv1.Duty) + require.NoError(t, SendReceive(ctx, client, server.ID(), req, resp, pID)) + + // Client sent the request and server sent the identical echoed response. + sentCount, sentSum := histSample(t, sentMsgSizeHist, string(pID)) + require.Equal(t, sentCount0+2, sentCount) + require.InDelta(t, sentSum0+2*msgSize, sentSum, 0.1) + + // Server received the request from the client. + recvServer, recvServerSum := histSample(t, receivedMsgSizeHist, string(pID), PeerName(client.ID())) + require.Equal(t, recvServer0+1, recvServer) + require.InDelta(t, msgSize, recvServerSum, 0.1) + + // Client received the response from the server. + recvClient, _ := histSample(t, receivedMsgSizeHist, string(pID), PeerName(server.ID())) + require.Equal(t, recvClient0+1, recvClient) +} + +func TestMessageReadErrorMetric(t *testing.T) { + var ( + pID = protocol.ID("test-read-limit") + ctx = context.Background() + client = testutil.CreateHost(t, testutil.AvailableAddr(t)) + server = testutil.CreateHost(t, testutil.AvailableAddr(t)) + ) + + client.Peerstore().AddAddrs(server.ID(), server.Addrs(), peerstore.PermanentAddrTTL) + + RegisterHandler("server", server, pID, + func() proto.Message { return new(pbv1.ParSigExMsg) }, + func(context.Context, peer.ID, proto.Message) (proto.Message, bool, error) { + require.Fail(t, "handler must not be called for oversized message") + return nil, false, nil + }, + WithReadLimit(16), + ) + + errCounter := msgReadErrorCounter.WithLabelValues(string(pID), PeerName(client.ID())) + errCount0 := promtestutil.ToFloat64(errCounter) + + // Message larger than the 16 byte read limit is rejected server side before the handler. + // Send is one-way so the local write succeeds regardless. + msg := &pbv1.ParSigExMsg{Duty: &pbv1.Duty{Slot: 99, Type: 2}, DataSet: &pbv1.ParSignedDataSet{ + Set: map[string]*pbv1.ParSignedData{"0xdeadbeef": {Data: make([]byte, 1024), Signature: make([]byte, 96)}}, + }} + require.NoError(t, Send(ctx, client, pID, server.ID(), msg)) + + require.Eventually(t, func() bool { + return promtestutil.ToFloat64(errCounter) >= errCount0+1 + }, time.Second*5, time.Millisecond*10) +} + +func TestInflightRequestMetrics(t *testing.T) { + var ( + pID = protocol.ID("test-inflight") + ctx = context.Background() + client = testutil.CreateHost(t, testutil.AvailableAddr(t)) + server = testutil.CreateHost(t, testutil.AvailableAddr(t)) + ) + + client.Peerstore().AddAddrs(server.ID(), server.Addrs(), peerstore.PermanentAddrTTL) + + RegisterHandler("server", server, pID, + func() proto.Message { return new(pbv1.Duty) }, + func(_ context.Context, _ peer.ID, req proto.Message) (proto.Message, bool, error) { + return req, true, nil + }, + ) + + clientName := PeerName(client.ID()) + + resp := new(pbv1.Duty) + require.NoError(t, SendReceive(ctx, client, server.ID(), &pbv1.Duty{Slot: 1}, resp, pID)) + + // One arrival observed in the concurrency histogram with count 1. + concCount, concSum := histSample(t, concurrentRequestsHist, string(pID), clientName) + require.Equal(t, uint64(1), concCount) + require.InDelta(t, 1, concSum, 0.1) + + // Handler duration observed once. + durCount, durSum := histSample(t, handlerDuration, string(pID)) + require.Equal(t, uint64(1), durCount) + require.Positive(t, durSum) + + // In-flight gauge back to zero after completion. + gauge, err := inflightGauge.GetMetricWithLabelValues(string(pID), clientName) + require.NoError(t, err) + require.Zero(t, promtestutil.ToFloat64(gauge)) +} diff --git a/p2p/receive.go b/p2p/receive.go index dc05e28c67..ae007b2cb5 100644 --- a/p2p/receive.go +++ b/p2p/receive.go @@ -53,6 +53,9 @@ func RegisterHandler(logTopic string, p2pNode host.Host, pID protocol.ID, t0 := time.Now() name := PeerName(s.Conn().RemotePeer()) + handlerDone := observeHandlerStart(s.Protocol(), s.Conn().RemotePeer()) + defer handlerDone() + _ = s.SetReadDeadline(time.Now().Add(o.receiveTimeout)) ctx, cancel := context.WithTimeout(context.Background(), o.receiveTimeout) ctx = log.WithTopic(ctx, logTopic) @@ -82,16 +85,22 @@ func RegisterHandler(logTopic string, p2pNode host.Host, pID protocol.ID, if IsRelayError(err) { return // Ignore relay errors. } else if netErr := net.Error(nil); errors.As(err, &netErr) && netErr.Timeout() { + incMessageReadError(s.Protocol(), s.Conn().RemotePeer()) log.Error(ctx, "Timeout reading p2p message from peer. This may indicate network latency issues or unresponsive peer", err, z.Any("duration", time.Since(t0))) + return } else if err != nil { + incMessageReadError(s.Protocol(), s.Conn().RemotePeer()) log.Error(ctx, "Failed to read p2p request from peer. Check network connectivity and peer health", err, z.Any("duration", time.Since(t0))) + return } else if err := protonil.Check(req); err != nil { log.Warn(ctx, "LibP2P received invalid proto", err) return } + observeReceivedMessage(s.Protocol(), s.Conn().RemotePeer(), req) + resp, ok, err := handlerFunc(ctx, s.Conn().RemotePeer(), req) if err != nil { log.Error(ctx, "P2P stream handler encountered an error. The request could not be processed", err, z.Any("duration", time.Since(t0))) @@ -108,5 +117,7 @@ func RegisterHandler(logTopic string, p2pNode host.Host, pID protocol.ID, log.Error(ctx, "Failed to write p2p response to peer. Connection may have been closed", err) return } + + observeSentMessage(s.Protocol(), resp) }) } diff --git a/p2p/sender.go b/p2p/sender.go index 3006c03fb9..c7fc55ec18 100644 --- a/p2p/sender.go +++ b/p2p/sender.go @@ -344,6 +344,8 @@ func SendReceive(ctx context.Context, p2pNode host.Host, peerID peer.ID, return errors.Wrap(err, "write request", z.Any("protocol", s.Protocol())) } + observeSentMessage(s.Protocol(), req) + if err := s.CloseWrite(); err != nil { // A canceled-stream error here is benign: the request was already written and // delivered above, and the peer resetting our send-direction (STOP_SENDING) does @@ -359,9 +361,12 @@ func SendReceive(ctx context.Context, p2pNode host.Host, peerID peer.ID, } if err = reader.ReadMsg(resp); err != nil { + incMessageReadError(s.Protocol(), peerID) return errors.Wrap(err, "read response", z.Any("protocol", s.Protocol())) } + observeReceivedMessage(s.Protocol(), peerID, resp) + o.rttCallback(time.Since(t0)) return nil @@ -408,6 +413,8 @@ func Send(ctx context.Context, p2pNode host.Host, protoID protocol.ID, peerID pe return errors.Wrap(err, "write message", z.Any("protocol", s.Protocol())) } + observeSentMessage(s.Protocol(), msg) + return nil } From 7ac84c563dd03f35f03d0d63f86791ed687b2ff8 Mon Sep 17 00:00:00 2001 From: kalo <24719519+KaloyanTanev@users.noreply.github.com> Date: Thu, 3 Sep 2026 16:29:28 +0300 Subject: [PATCH 2/7] p2p: update inflight gauge under lock --- p2p/metrics.go | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/p2p/metrics.go b/p2p/metrics.go index 0694c32457..8910c5f88d 100644 --- a/p2p/metrics.go +++ b/p2p/metrics.go @@ -160,18 +160,18 @@ func observeHandlerStart(pID protocol.ID, peerID peer.ID) func() { inflightCounts.Lock() inflightCounts.counts[key]++ n := inflightCounts.counts[key] + // Update the gauge while holding the lock so concurrent Set calls cannot be reordered. + inflightGauge.WithLabelValues(key[0], key[1]).Set(float64(n)) inflightCounts.Unlock() - inflightGauge.WithLabelValues(key[0], key[1]).Set(float64(n)) concurrentRequestsHist.WithLabelValues(key[0], key[1]).Observe(float64(n)) return func() { inflightCounts.Lock() inflightCounts.counts[key]-- - n := inflightCounts.counts[key] + inflightGauge.WithLabelValues(key[0], key[1]).Set(float64(inflightCounts.counts[key])) inflightCounts.Unlock() - inflightGauge.WithLabelValues(key[0], key[1]).Set(float64(n)) handlerDuration.WithLabelValues(key[0]).Observe(time.Since(t0).Seconds()) } } From 1f00c409ea2161bfa5d28d024f02791713e54b3d Mon Sep 17 00:00:00 2001 From: kalo <24719519+KaloyanTanev@users.noreply.github.com> Date: Thu, 3 Sep 2026 16:40:13 +0300 Subject: [PATCH 3/7] p2p: cleanup idle inflight keys, observe sent after close write --- p2p/metrics.go | 6 +++++- p2p/sender.go | 4 ++-- 2 files changed, 7 insertions(+), 3 deletions(-) diff --git a/p2p/metrics.go b/p2p/metrics.go index 8910c5f88d..c9712b031e 100644 --- a/p2p/metrics.go +++ b/p2p/metrics.go @@ -169,7 +169,11 @@ func observeHandlerStart(pID protocol.ID, peerID peer.ID) func() { return func() { inflightCounts.Lock() inflightCounts.counts[key]-- - inflightGauge.WithLabelValues(key[0], key[1]).Set(float64(inflightCounts.counts[key])) + n := inflightCounts.counts[key] + if n <= 0 { + delete(inflightCounts.counts, key) // Don't grow the map with idle keys. + } + inflightGauge.WithLabelValues(key[0], key[1]).Set(float64(n)) inflightCounts.Unlock() handlerDuration.WithLabelValues(key[0]).Observe(time.Since(t0).Seconds()) diff --git a/p2p/sender.go b/p2p/sender.go index c7fc55ec18..b5b96c82d3 100644 --- a/p2p/sender.go +++ b/p2p/sender.go @@ -344,8 +344,6 @@ func SendReceive(ctx context.Context, p2pNode host.Host, peerID peer.ID, return errors.Wrap(err, "write request", z.Any("protocol", s.Protocol())) } - observeSentMessage(s.Protocol(), req) - if err := s.CloseWrite(); err != nil { // A canceled-stream error here is benign: the request was already written and // delivered above, and the peer resetting our send-direction (STOP_SENDING) does @@ -360,6 +358,8 @@ func SendReceive(ctx context.Context, p2pNode host.Host, peerID peer.ID, } } + observeSentMessage(s.Protocol(), req) + if err = reader.ReadMsg(resp); err != nil { incMessageReadError(s.Protocol(), peerID) return errors.Wrap(err, "read response", z.Any("protocol", s.Protocol())) From b0dfc5dd9ec25b4dc51fab094e4e8681044ec3ed Mon Sep 17 00:00:00 2001 From: kalo <24719519+KaloyanTanev@users.noreply.github.com> Date: Thu, 3 Sep 2026 16:56:53 +0300 Subject: [PATCH 4/7] p2p: consistent relay error handling, concurrency test --- p2p/metrics_internal_test.go | 59 ++++++++++++++++++++++++++++++++++++ p2p/sender.go | 6 +++- 2 files changed, 64 insertions(+), 1 deletion(-) diff --git a/p2p/metrics_internal_test.go b/p2p/metrics_internal_test.go index e7e25492b7..79a9217c26 100644 --- a/p2p/metrics_internal_test.go +++ b/p2p/metrics_internal_test.go @@ -4,6 +4,7 @@ package p2p import ( "context" + "sync" "testing" "time" @@ -147,3 +148,61 @@ func TestInflightRequestMetrics(t *testing.T) { require.NoError(t, err) require.Zero(t, promtestutil.ToFloat64(gauge)) } + +func TestConcurrentRequestDepths(t *testing.T) { + var ( + pID = protocol.ID("test-concurrency") + ctx = context.Background() + client = testutil.CreateHost(t, testutil.AvailableAddr(t)) + server = testutil.CreateHost(t, testutil.AvailableAddr(t)) + ) + + client.Peerstore().AddAddrs(server.ID(), server.Addrs(), peerstore.PermanentAddrTTL) + + release := make(chan struct{}) + RegisterHandler("server", server, pID, + func() proto.Message { return new(pbv1.Duty) }, + func(_ context.Context, _ peer.ID, req proto.Message) (proto.Message, bool, error) { + <-release + return req, true, nil + }, + ) + + clientName := PeerName(client.ID()) + + // Issue concurrent requests against a blocked handler. + const n = 3 + + var wg sync.WaitGroup + for range n { + wg.Add(1) + + go func() { + defer wg.Done() + + resp := new(pbv1.Duty) + _ = SendReceive(ctx, client, server.ID(), &pbv1.Duty{Slot: 1}, resp, pID) + }() + } + + gauge, err := inflightGauge.GetMetricWithLabelValues(string(pID), clientName) + require.NoError(t, err) + + // All requests block in the handler, so the gauge reaches n. + require.Eventually(t, func() bool { + return promtestutil.ToFloat64(gauge) == n + }, time.Second*5, time.Millisecond*10) + + close(release) + wg.Wait() + + // Gauge returns to zero once all handlers complete. + require.Eventually(t, func() bool { + return promtestutil.ToFloat64(gauge) == 0 + }, time.Second*5, time.Millisecond*10) + + // The concurrency histogram observed depths 1, 2 and 3 (in some order), so the sum is 6. + concCount, concSum := histSample(t, concurrentRequestsHist, string(pID), clientName) + require.Equal(t, uint64(n), concCount) + require.InDelta(t, 6, concSum, 0.1) +} diff --git a/p2p/sender.go b/p2p/sender.go index b5b96c82d3..06ca022237 100644 --- a/p2p/sender.go +++ b/p2p/sender.go @@ -361,7 +361,11 @@ func SendReceive(ctx context.Context, p2pNode host.Host, peerID peer.ID, observeSentMessage(s.Protocol(), req) if err = reader.ReadMsg(resp); err != nil { - incMessageReadError(s.Protocol(), peerID) + // Relay resets are benign churn, exclude them like the receive path does. + if !IsRelayError(err) { + incMessageReadError(s.Protocol(), peerID) + } + return errors.Wrap(err, "read response", z.Any("protocol", s.Protocol())) } From b0d121ab69071e21fa6410473f1eeec637429a9d Mon Sep 17 00:00:00 2001 From: kalo <24719519+KaloyanTanev@users.noreply.github.com> Date: Thu, 3 Sep 2026 17:12:33 +0300 Subject: [PATCH 5/7] p2p: count direct-stream resets as read errors --- p2p/receive.go | 6 ++++++ p2p/sender.go | 5 +++-- 2 files changed, 9 insertions(+), 2 deletions(-) diff --git a/p2p/receive.go b/p2p/receive.go index ae007b2cb5..a5a6f9a69d 100644 --- a/p2p/receive.go +++ b/p2p/receive.go @@ -83,6 +83,12 @@ func RegisterHandler(logTopic string, p2pNode host.Host, pID protocol.ID, err := readFunc(s).ReadMsg(req) if IsRelayError(err) { + // Resets on relayed (limited) connections are benign circuit recycling, + // but a reset on a direct connection is a genuine read failure. + if !s.Conn().Stat().Limited { + incMessageReadError(s.Protocol(), s.Conn().RemotePeer()) + } + return // Ignore relay errors. } else if netErr := net.Error(nil); errors.As(err, &netErr) && netErr.Timeout() { incMessageReadError(s.Protocol(), s.Conn().RemotePeer()) diff --git a/p2p/sender.go b/p2p/sender.go index 06ca022237..c0a1a3d935 100644 --- a/p2p/sender.go +++ b/p2p/sender.go @@ -361,8 +361,9 @@ func SendReceive(ctx context.Context, p2pNode host.Host, peerID peer.ID, observeSentMessage(s.Protocol(), req) if err = reader.ReadMsg(resp); err != nil { - // Relay resets are benign churn, exclude them like the receive path does. - if !IsRelayError(err) { + // Resets on relayed (limited) connections are benign circuit recycling, + // but any other failure (including a reset on a direct connection) counts. + if !IsRelayError(err) || !s.Conn().Stat().Limited { incMessageReadError(s.Protocol(), peerID) } From 25be7e8700be914bcc156290b51031cfbfe713a9 Mon Sep 17 00:00:00 2001 From: kalo <24719519+KaloyanTanev@users.noreply.github.com> Date: Thu, 3 Sep 2026 17:19:32 +0300 Subject: [PATCH 6/7] p2p: observe received size before proto validation --- p2p/receive.go | 10 +++++++--- 1 file changed, 7 insertions(+), 3 deletions(-) diff --git a/p2p/receive.go b/p2p/receive.go index a5a6f9a69d..69b9635111 100644 --- a/p2p/receive.go +++ b/p2p/receive.go @@ -99,14 +99,18 @@ func RegisterHandler(logTopic string, p2pNode host.Host, pID protocol.ID, incMessageReadError(s.Protocol(), s.Conn().RemotePeer()) log.Error(ctx, "Failed to read p2p request from peer. Check network connectivity and peer health", err, z.Any("duration", time.Since(t0))) - return - } else if err := protonil.Check(req); err != nil { - log.Warn(ctx, "LibP2P received invalid proto", err) return } + // Observe before application-level validation so malformed-but-readable messages + // are still visible in the size histogram. observeReceivedMessage(s.Protocol(), s.Conn().RemotePeer(), req) + if err := protonil.Check(req); err != nil { + log.Warn(ctx, "LibP2P received invalid proto", err) + return + } + resp, ok, err := handlerFunc(ctx, s.Conn().RemotePeer(), req) if err != nil { log.Error(ctx, "P2P stream handler encountered an error. The request could not be processed", err, z.Any("duration", time.Since(t0))) From 59ad34f6f28f705192becaad949d88a681130371 Mon Sep 17 00:00:00 2001 From: kalo <24719519+KaloyanTanev@users.noreply.github.com> Date: Thu, 3 Sep 2026 17:51:39 +0300 Subject: [PATCH 7/7] p2p: named struct for inflight key --- p2p/metrics.go | 22 +++++++++++++++------- 1 file changed, 15 insertions(+), 7 deletions(-) diff --git a/p2p/metrics.go b/p2p/metrics.go index c9712b031e..96e80765d2 100644 --- a/p2p/metrics.go +++ b/p2p/metrics.go @@ -143,40 +143,48 @@ var ( }, []string{"protocol"}) ) +// inflightKey identifies a per-protocol, per-peer inflight request counter. +type inflightKey struct { + protocol string + peer string +} + // inflightCounts tracks the number of concurrently handled inbound messages per (protocol, peer). // It is the source of truth for both inflightGauge and concurrentRequestsHist so they never drift. var inflightCounts = struct { sync.Mutex - counts map[[2]string]int -}{counts: make(map[[2]string]int)} + counts map[inflightKey]int +}{counts: make(map[inflightKey]int)} // observeHandlerStart records the start of inbound message handling and returns a done function // to be called (deferred) when handling completes. func observeHandlerStart(pID protocol.ID, peerID peer.ID) func() { - key := [2]string{string(pID), PeerName(peerID)} + key := inflightKey{protocol: string(pID), peer: PeerName(peerID)} t0 := time.Now() inflightCounts.Lock() inflightCounts.counts[key]++ n := inflightCounts.counts[key] // Update the gauge while holding the lock so concurrent Set calls cannot be reordered. - inflightGauge.WithLabelValues(key[0], key[1]).Set(float64(n)) + inflightGauge.WithLabelValues(key.protocol, key.peer).Set(float64(n)) inflightCounts.Unlock() - concurrentRequestsHist.WithLabelValues(key[0], key[1]).Observe(float64(n)) + concurrentRequestsHist.WithLabelValues(key.protocol, key.peer).Observe(float64(n)) return func() { inflightCounts.Lock() inflightCounts.counts[key]-- + n := inflightCounts.counts[key] if n <= 0 { delete(inflightCounts.counts, key) // Don't grow the map with idle keys. } - inflightGauge.WithLabelValues(key[0], key[1]).Set(float64(n)) + + inflightGauge.WithLabelValues(key.protocol, key.peer).Set(float64(n)) inflightCounts.Unlock() - handlerDuration.WithLabelValues(key[0]).Observe(time.Since(t0).Seconds()) + handlerDuration.WithLabelValues(key.protocol).Observe(time.Since(t0).Seconds()) } }