p2p: fix priority queue bytes pending calculation (#7120)

This metric describes itself as 'pending' but never actual decrements when the messages are removed from the queue.

This change fixes that by decrementing the metric when the data is removed from the queue.

internal/p2p/pqueue.go

@@ -167,13 +167,12 @@ func (s *pqScheduler) process() {
 				timestamp: time.Now().UTC(),
 			}
 
-			s.metrics.PeerPendingSendBytes.With("peer_id", string(pqEnv.envelope.To)).Add(float64(pqEnv.size))
-
 			// enqueue
 
 			// Check if we have sufficient capacity to simply enqueue the incoming
 			// Envelope.
 			if s.size+pqEnv.size <= s.capacity {
+				s.metrics.PeerPendingSendBytes.With("peer_id", string(pqEnv.envelope.To)).Add(float64(pqEnv.size))
 				// enqueue the incoming Envelope
 				s.push(pqEnv)
 			} else {
@@ -213,6 +212,8 @@ func (s *pqScheduler) process() {
 									"capacity", s.capacity,
 								)
 
+								s.metrics.PeerPendingSendBytes.With("peer_id", string(pqEnvTmp.envelope.To)).Add(float64(-pqEnvTmp.size))
+
 								// dequeue/drop from the priority queue
 								heap.Remove(s.pq, pqEnvTmp.index)
 
@@ -257,6 +258,8 @@ func (s *pqScheduler) process() {
 				s.metrics.PeerSendBytesTotal.With(
 					"chID", chIDStr,
 					"peer_id", string(pqEnv.envelope.To)).Add(float64(pqEnv.size))
+				s.metrics.PeerPendingSendBytes.With(
+					"peer_id", string(pqEnv.envelope.To)).Add(float64(-pqEnv.size))
 				select {
 				case s.dequeueCh <- pqEnv.envelope:
 				case <-s.closer.Done():